This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new c3278a5 feat: add Pipeline domain runtime (#177)
c3278a5 is described below
commit c3278a5026881d2934571f6cfb9be3a4ab58f0fc
Author: Yijia Su <[email protected]>
AuthorDate: Fri Jul 31 18:17:15 2026 +0800
feat: add Pipeline domain runtime (#177)
Adds the read-only Pipeline domain with capability detection, isolated
probes, formal handlers, tests, and changelog coverage.
---
CHANGELOG.md | 6 +
doris_mcp_server/tools/capability_detector.py | 195 ++-
doris_mcp_server/tools/domain_catalog.py | 20 +-
doris_mcp_server/tools/domain_dispatcher.py | 27 +-
doris_mcp_server/tools/pipeline_handlers.py | 109 ++
doris_mcp_server/tools/tools_manager.py | 3 +
doris_mcp_server/utils/pipeline_runtime.py | 1958 ++++++++++++++++++++++++
test/integration/test_real_doris_transports.py | 122 ++
test/tools/test_capability_detector.py | 82 +-
test/tools/test_domain_dispatcher.py | 57 +-
test/utils/test_pipeline_runtime.py | 348 +++++
11 files changed, 2906 insertions(+), 21 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 75cd49e..7ec61bb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -51,6 +51,9 @@ under **Unreleased** until a new version is selected and
published.
- A read-only Cluster domain with eleven capability-gated children for node
inventory, active tasks, metrics, memory, cache, compaction, workload and
compute groups, recorded resource growth, and sanitized runtime evidence.
+- A read-only Pipeline domain with five capability-gated children for
+ ingestion status and diagnosis, materialized-view refresh state, recorded
+ table freshness, and bounded upstream or downstream dependency evidence.
- Real Doris process tests covering Streamable HTTP and stdio.
### Changed
@@ -92,6 +95,9 @@ under **Unreleased** until a new version is selected and
published.
- Released failed capability-probe connections from their captured owner pools
so a single-connection route cannot starve subsequent domain calls.
+- Isolated Doris domain probe statements in independent route-aware connection
+ contexts so one unsupported version-specific statement cannot poison later
+ capability evidence.
- Excluded explicitly dead Doris components from active version gating while
preserving them in node inventory, and kept live runtime manifests within
the 16 KiB domain budget.
diff --git a/doris_mcp_server/tools/capability_detector.py
b/doris_mcp_server/tools/capability_detector.py
index 7d78779..467c4e5 100644
--- a/doris_mcp_server/tools/capability_detector.py
+++ b/doris_mcp_server/tools/capability_detector.py
@@ -216,6 +216,45 @@ _DOMAIN_PROBES: Mapping[str, tuple[tuple[str, tuple[str,
...]], ...]] = {
("metrics_history_readable",),
),
),
+ "doris_pipeline": (
+ (
+ "SHOW LOAD LIMIT 1",
+ ("batch_load_metadata_readable",),
+ ),
+ (
+ "SHOW STREAM LOAD LIMIT 1",
+ ("stream_load_metadata_readable",),
+ ),
+ (
+ "SHOW ALL ROUTINE LOAD",
+ ("routine_load_metadata_readable",),
+ ),
+ (
+ 'SELECT Id, Name, Status FROM jobs("type"="insert") LIMIT 1',
+ (
+ "insert_job_metadata_readable",
+ "continuous_load_metadata_readable",
+ ),
+ ),
+ (
+ 'SELECT Id, Name, Status FROM jobs("type"="mv") LIMIT 1',
+ ("materialized_view_job_metadata_readable",),
+ ),
+ (
+ 'SELECT TaskId, Status FROM tasks("type"="mv") LIMIT 1',
+ ("materialized_view_task_metadata_readable",),
+ ),
+ (
+ (
+ "SELECT `time`, stmt "
+ "FROM internal.__internal_schema.audit_log LIMIT 1"
+ ),
+ (
+ "pipeline_audit_log_readable",
+ "audit_log_readable",
+ ),
+ ),
+ ),
}
@@ -306,32 +345,33 @@ class DorisCapabilityDetector:
"Doris route changed before domain capability probing"
)
- session_id = (
-
f"capability-domain:{domain_name}:{requested_route.fingerprint[:12]}"
- )
async with asyncio.timeout(self._probe_timeout_seconds):
- async with (
-
self._connection_manager.get_connection_context_for_auth_context(
- session_id,
- auth_context,
- ) as connection
+ probes = dict(base.probes)
+ for index, (sql, probe_ids) in enumerate(
+ _DOMAIN_PROBES.get(domain_name, ())
):
current_route = self.route_identity(auth_context)
if current_route.fingerprint != base.route.fingerprint:
raise CapabilityRouteChangedError(
"Doris route changed during domain capability
probing"
)
- probes = dict(base.probes)
- for sql, probe_ids in _DOMAIN_PROBES.get(
- domain_name,
- (),
+ session_id = (
+ f"capability-domain:{domain_name}:{index}:"
+ f"{requested_route.fingerprint[:12]}"
+ )
+ async with (
+ self._connection_manager
+ .get_connection_context_for_auth_context(
+ session_id,
+ auth_context,
+ ) as connection
):
evidence = await self._probe_statement(
connection,
sql,
probe_ids,
)
- probes.update(evidence)
+ probes.update(evidence)
if domain_name == "doris_query":
probes.update(await
self._probe_query_services(auth_context))
probes["profile_or_audit_readable"] =
_combine_query_evidence_probe(
@@ -339,6 +379,8 @@ class DorisCapabilityDetector:
)
elif domain_name == "doris_cluster":
probes.update(await
self._safe_probe_cluster_services(auth_context))
+ elif domain_name == "doris_pipeline":
+ probes.update(_combine_pipeline_evidence_probes(probes))
completed_route = self.route_identity(auth_context)
if completed_route.fingerprint != base.route.fingerprint:
raise CapabilityRouteChangedError(
@@ -1264,6 +1306,133 @@ def _combine_query_evidence_probe(
)
+def _combine_pipeline_evidence_probes(
+ probes: Mapping[str, CapabilityProbeEvidence],
+) -> dict[str, CapabilityProbeEvidence]:
+ ingestion_source_ids = (
+ "batch_load_metadata_readable",
+ "stream_load_metadata_readable",
+ "routine_load_metadata_readable",
+ )
+ all_ingestion_source_ids = (
+ *ingestion_source_ids,
+ "insert_job_metadata_readable",
+ )
+ ingestion = _combine_any_runtime_probe(
+ "stream_broker_routine_load_readable",
+ probes,
+ ingestion_source_ids,
+ supported_reason="LOAD_JOB_METADATA_READABLE",
+ )
+ ingestion_status = _combine_any_runtime_probe(
+ "ingestion_status_readable",
+ probes,
+ all_ingestion_source_ids,
+ supported_reason="INGESTION_STATUS_READABLE",
+ )
+ materialized_views = _combine_any_runtime_probe(
+ "materialized_view_tasks_readable",
+ probes,
+ (
+ "materialized_view_job_metadata_readable",
+ "materialized_view_task_metadata_readable",
+ ),
+ supported_reason="MATERIALIZED_VIEW_METADATA_READABLE",
+ )
+
+ freshness_candidates = tuple(
+ probe
+ for probe_id in (
+ "routine_load_metadata_readable",
+ "pipeline_audit_log_readable",
+ )
+ if (probe := probes.get(probe_id)) is not None
+ )
+ if any(
+ probe.probe_id == "routine_load_metadata_readable"
+ and probe.status is CapabilityProbeStatus.SUPPORTED
+ for probe in freshness_candidates
+ ):
+ freshness_status = CapabilityProbeStatus.SUPPORTED
+ freshness_reason = "LOAD_OFFSET_OR_TIME_EVIDENCE_READABLE"
+ elif any(
+ probe.status is CapabilityProbeStatus.SUPPORTED
+ for probe in freshness_candidates
+ ):
+ freshness_status = CapabilityProbeStatus.DEGRADED
+ freshness_reason = "AUDIT_TIME_FALLBACK_READABLE"
+ else:
+ freshness_status, freshness_reason = _best_candidate_status(
+ freshness_candidates,
+ reason_prefix="FRESHNESS_EVIDENCE",
+ )
+ freshness = CapabilityProbeEvidence(
+ probe_id="freshness_evidence_readable",
+ status=freshness_status,
+ reason_code=freshness_reason,
+ evidence_sources=("runtime_probe",),
+ )
+ load_offset = CapabilityProbeEvidence(
+ probe_id="load_offset_or_time_evidence_readable",
+ status=freshness_status,
+ reason_code=freshness_reason,
+ evidence_sources=("runtime_probe",),
+ )
+ return {
+ ingestion.probe_id: ingestion,
+ ingestion_status.probe_id: ingestion_status,
+ materialized_views.probe_id: materialized_views,
+ freshness.probe_id: freshness,
+ load_offset.probe_id: load_offset,
+ }
+
+
+def _combine_any_runtime_probe(
+ probe_id: str,
+ probes: Mapping[str, CapabilityProbeEvidence],
+ candidate_ids: Sequence[str],
+ *,
+ supported_reason: str,
+) -> CapabilityProbeEvidence:
+ candidates = tuple(
+ probe
+ for candidate_id in candidate_ids
+ if (probe := probes.get(candidate_id)) is not None
+ )
+ if any(
+ probe.status is CapabilityProbeStatus.SUPPORTED for probe in candidates
+ ):
+ status = CapabilityProbeStatus.SUPPORTED
+ reason = supported_reason
+ else:
+ status, reason = _best_candidate_status(
+ candidates,
+ reason_prefix=probe_id.upper(),
+ )
+ return CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=status,
+ reason_code=reason,
+ evidence_sources=("runtime_probe",),
+ )
+
+
+def _best_candidate_status(
+ candidates: Sequence[CapabilityProbeEvidence],
+ *,
+ reason_prefix: str,
+) -> tuple[CapabilityProbeStatus, str]:
+ for status in (
+ CapabilityProbeStatus.DEGRADED,
+ CapabilityProbeStatus.MISCONFIGURED,
+ CapabilityProbeStatus.UNKNOWN,
+ CapabilityProbeStatus.UNSUPPORTED,
+ ):
+ if any(probe.status is status for probe in candidates):
+ return status, f"{reason_prefix}_{status.value.upper()}"
+ return CapabilityProbeStatus.UNKNOWN, f"{reason_prefix}_UNKNOWN"
+
+
def _row_value(
row: Mapping[str, Any],
*names: str,
diff --git a/doris_mcp_server/tools/domain_catalog.py
b/doris_mcp_server/tools/domain_catalog.py
index 63ce689..8c55845 100644
--- a/doris_mcp_server/tools/domain_catalog.py
+++ b/doris_mcp_server/tools/domain_catalog.py
@@ -1332,11 +1332,24 @@ DOMAIN_DEFINITIONS = (
"Read streaming and batch load job status.",
_input_schema(
{
- "job_types": _string_array("Load job types."),
+ "job_types": _string_array(
+ "Ingestion job types.",
+ enum=(
+ "batch_load",
+ "stream_load",
+ "routine_load",
+ "insert_job",
+ "continuous_load",
+ ),
+ ),
"database": _string("Database filter."),
"table": _string("Table filter."),
"states": _string_array("Job states."),
- "limit": _integer("Maximum results.", minimum=1),
+ "limit": _integer(
+ "Maximum results.",
+ minimum=1,
+ maximum=200,
+ ),
}
),
_COLLECTION_OUTPUT,
@@ -1355,6 +1368,7 @@ DOMAIN_DEFINITIONS = (
"window_minutes": _integer(
"Lookback window in minutes.",
minimum=1,
+ maximum=43_200,
),
"include_dependencies": _boolean(
"Include dependency evidence."
@@ -1397,6 +1411,7 @@ DOMAIN_DEFINITIONS = (
"threshold_seconds": _integer(
"Freshness threshold in seconds.",
minimum=1,
+ maximum=31_536_000,
),
"time_column": _string("Optional event-time column."),
},
@@ -1421,6 +1436,7 @@ DOMAIN_DEFINITIONS = (
"depth": _integer(
"Maximum traversal depth.",
minimum=1,
+ maximum=5,
),
},
required=("object",),
diff --git a/doris_mcp_server/tools/domain_dispatcher.py
b/doris_mcp_server/tools/domain_dispatcher.py
index 0d8b60b..41e1ae3 100644
--- a/doris_mcp_server/tools/domain_dispatcher.py
+++ b/doris_mcp_server/tools/domain_dispatcher.py
@@ -45,6 +45,7 @@ from ..state_handles import StateHandleCodec, StateHandleError
from ..utils.catalog_metadata import CatalogMetadataFailure
from ..utils.cluster_runtime import ClusterRuntimeFailure
from ..utils.logger import get_audit_logger, get_logger
+from ..utils.pipeline_runtime import PipelineRuntimeFailure
from ..utils.query_runtime import QueryRuntimeFailure
from . import domain_catalog as domain_catalog_module
from .domain_catalog import (
@@ -168,7 +169,8 @@ def _build_bindings(
for domain in catalog.domains:
for child in domain.children:
feature_id = f"{domain.name}.{child.name}"
- if feature_id in bindings:
+ existing = bindings.get(feature_id)
+ if isinstance(existing, tuple):
continue
handler_name = _formal_handler_name(domain.name, child.name)
if callable(getattr(owner, handler_name, None)):
@@ -660,6 +662,29 @@ class DomainDispatcher:
"status_code": exc.status_code,
},
)
+ except PipelineRuntimeFailure as exc:
+ self._audit(feature_id, arguments, "error", started)
+ return self._error(
+ domain.name,
+ (
+ DomainErrorCode.CHILD_ARGUMENTS_INVALID
+ if exc.reason_code
+ in {
+ "PIPELINE_ARGUMENT_INVALID",
+ "PIPELINE_DATABASE_REQUIRED",
+ }
+ else DomainErrorCode.CHILD_EXECUTION_FAILED
+ ),
+ str(exc),
+ child_tool=child.name,
+ manifest_version=manifest.manifest_version,
+ retryable=exc.retryable,
+ details={
+ "rediscover": False,
+ "reason_code": exc.reason_code,
+ "status_code": exc.status_code,
+ },
+ )
except ToolOutputValidationError:
logger.exception("Formal child output validation failed for %s",
feature_id)
self._audit(
diff --git a/doris_mcp_server/tools/pipeline_handlers.py
b/doris_mcp_server/tools/pipeline_handlers.py
new file mode 100644
index 0000000..c091110
--- /dev/null
+++ b/doris_mcp_server/tools/pipeline_handlers.py
@@ -0,0 +1,109 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Formal Pipeline-domain handlers backed by one strict runtime."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, cast
+
+from ..utils.db import DorisConnectionManager
+from ..utils.pipeline_runtime import DorisPipelineRuntime
+
+
+class _PipelineHandlerOwner(Protocol):
+ """State supplied by ``DorisToolsManager`` to the mixin."""
+
+ pipeline_runtime: DorisPipelineRuntime
+
+
+class PipelineToolHandlersMixin:
+ """Route every Pipeline child through the evidence-bearing runtime."""
+
+ def _initialize_pipeline_handlers(
+ self: _PipelineHandlerOwner,
+ connection_manager: DorisConnectionManager,
+ ) -> None:
+ self.pipeline_runtime = DorisPipelineRuntime(connection_manager)
+
+ async def _formal_doris_pipeline_get_ingestion_status_tool(
+ self: _PipelineHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.pipeline_runtime.get_ingestion_status(
+ job_types=cast(list[str] | None, arguments.get("job_types")),
+ database=cast(str | None, arguments.get("database")),
+ table=cast(str | None, arguments.get("table")),
+ states=cast(list[str] | None, arguments.get("states")),
+ limit=cast(int | None, arguments.get("limit")),
+ )
+
+ async def _formal_doris_pipeline_diagnose_ingestion_tool(
+ self: _PipelineHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.pipeline_runtime.diagnose_ingestion(
+ job_id=cast(str | None, arguments.get("job_id")),
+ database=cast(str | None, arguments.get("database")),
+ table=cast(str | None, arguments.get("table")),
+ window_minutes=cast(int | None, arguments.get("window_minutes")),
+ include_dependencies=bool(
+ arguments.get("include_dependencies", True)
+ ),
+ )
+
+ async def _formal_doris_pipeline_get_materialized_view_status_tool(
+ self: _PipelineHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.pipeline_runtime.get_materialized_view_status(
+ database=cast(str | None, arguments.get("database")),
+ view=cast(str | None, arguments.get("view")),
+ states=cast(list[str] | None, arguments.get("states")),
+ include_refresh_history=bool(
+ arguments.get("include_refresh_history", False)
+ ),
+ )
+
+ async def _formal_doris_pipeline_monitor_data_freshness_tool(
+ self: _PipelineHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.pipeline_runtime.monitor_data_freshness(
+ database=cast(str, arguments.get("database")),
+ table=cast(str, arguments.get("table")),
+ threshold_seconds=cast(
+ int | None,
+ arguments.get("threshold_seconds"),
+ ),
+ time_column=cast(str | None, arguments.get("time_column")),
+ )
+
+ async def _formal_doris_pipeline_analyze_data_dependencies_tool(
+ self: _PipelineHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.pipeline_runtime.analyze_data_dependencies(
+ catalog=cast(str | None, arguments.get("catalog")),
+ database=cast(str | None, arguments.get("database")),
+ object_name=cast(str, arguments.get("object")),
+ direction=cast(str | None, arguments.get("direction")),
+ depth=cast(int | None, arguments.get("depth")),
+ )
+
+
+__all__ = ["PipelineToolHandlersMixin"]
diff --git a/doris_mcp_server/tools/tools_manager.py
b/doris_mcp_server/tools/tools_manager.py
index 246903c..78dfa73 100644
--- a/doris_mcp_server/tools/tools_manager.py
+++ b/doris_mcp_server/tools/tools_manager.py
@@ -63,6 +63,7 @@ from .domain_manifest import (
DomainManifestService,
)
from .doris_feature_matrix import DORIS_FEATURE_MATRIX
+from .pipeline_handlers import PipelineToolHandlersMixin
from .query_handlers import QueryToolHandlersMixin
from .tool_provider import CustomToolProvider, ToolProviderRuntime
from .tool_registry import ToolRegistryError
@@ -74,6 +75,7 @@ class DorisToolsManager(
QueryToolHandlersMixin,
CatalogToolHandlersMixin,
ClusterToolHandlersMixin,
+ PipelineToolHandlersMixin,
DomainManifestManagerMixin,
):
"""Apache Doris Tools Manager"""
@@ -101,6 +103,7 @@ class DorisToolsManager(
connection_manager,
self.monitoring_tools,
)
+ self._initialize_pipeline_handlers(connection_manager)
# Initialize v0.5.0 advanced analytics tools
self.data_governance_tools = DataGovernanceTools(connection_manager)
diff --git a/doris_mcp_server/utils/pipeline_runtime.py
b/doris_mcp_server/utils/pipeline_runtime.py
new file mode 100644
index 0000000..641cc16
--- /dev/null
+++ b/doris_mcp_server/utils/pipeline_runtime.py
@@ -0,0 +1,1958 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Strict, evidence-bearing runtime for the read-only Pipeline domain."""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from collections import deque
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any
+
+import sqlparse
+from sqlparse import tokens as sql_tokens
+
+from .db import DorisConnectionManager
+from .redaction import redact_sensitive_data
+from .security import get_current_auth_context
+from .sql_security_utils import (
+ SQLSecurityError,
+ build_table_reference,
+ quote_identifier,
+ validate_identifier,
+)
+
+_MAX_BYTES = 2 * 1024 * 1024
+_MAX_LIMIT = 200
+_MAX_SOURCE_ROWS = 500
+_MAX_PARTITIONS = 1_000
+_MAX_DEPENDENCY_DEPTH = 5
+_MAX_DEPENDENCY_EDGES = 500
+_DEFAULT_FRESHNESS_THRESHOLD_SECONDS = 86_400
+_DEFAULT_DIAGNOSIS_WINDOW_MINUTES = 60
+_DEFAULT_DEPENDENCY_WINDOW_DAYS = 30
+_JOB_TYPES = frozenset(
+ {
+ "batch_load",
+ "stream_load",
+ "routine_load",
+ "insert_job",
+ "continuous_load",
+ }
+)
+_REFERENCE_IDENTIFIER = (
+ r"(?:`[^`]+`|[A-Za-z_][A-Za-z0-9_$]*)"
+ r"(?:\s*\.\s*(?:`[^`]+`|[A-Za-z_][A-Za-z0-9_$]*)){0,2}"
+)
+_SOURCE_REFERENCE = re.compile(
+ rf"\b(?:FROM|JOIN)\s+({_REFERENCE_IDENTIFIER})",
+ re.IGNORECASE,
+)
+_WRITE_TARGET = re.compile(
+ rf"\b(?:INSERT\s+INTO|CREATE\s+(?:MATERIALIZED\s+)?VIEW|"
+ rf"CREATE\s+TABLE)\s+({_REFERENCE_IDENTIFIER})",
+ re.IGNORECASE,
+)
+_CTE_NAME = re.compile(
+ r"(?:\bWITH|,)\s*(`[^`]+`|[A-Za-z_][A-Za-z0-9_$]*)\s+AS\s*\(",
+ re.IGNORECASE,
+)
+_ERROR_STATES = frozenset(
+ {
+ "cancelled",
+ "canceled",
+ "failed",
+ "fail",
+ "paused",
+ "stopped",
+ "error",
+ }
+)
+_RUNNING_STATES = frozenset(
+ {
+ "pending",
+ "etl",
+ "loading",
+ "running",
+ "need_schedule",
+ "waiting_txn",
+ }
+)
+_SUCCESS_STATES = frozenset({"finished", "success", "visible", "succeeded"})
+
+
+class PipelineRuntimeFailure(RuntimeError):
+ """Sanitized Pipeline-domain failure with a stable reason code."""
+
+ def __init__(
+ self,
+ message: str,
+ *,
+ reason_code: str,
+ status_code: int,
+ retryable: bool = False,
+ ) -> None:
+ super().__init__(message)
+ self.reason_code = reason_code
+ self.status_code = status_code
+ self.retryable = retryable
+
+
+class DorisPipelineRuntime:
+ """Read bounded ingestion, freshness, MV, and dependency evidence."""
+
+ def __init__(self, connection_manager: DorisConnectionManager) -> None:
+ self._connection_manager = connection_manager
+ self._session_prefix = f"formal_pipeline_{uuid.uuid4().hex[:8]}"
+
+ async def get_ingestion_status(
+ self,
+ *,
+ job_types: Sequence[str] | None,
+ database: str | None,
+ table: str | None,
+ states: Sequence[str] | None,
+ limit: int | None,
+ ) -> dict[str, Any]:
+ selected_types = _job_types(job_types)
+ selected_states = _normalized_filter(states)
+ result_limit = _bounded_int(limit, default=100, minimum=1,
maximum=_MAX_LIMIT)
+ database_name = await self._resolve_database(database)
+ table_name = _optional_identifier(table, "table name")
+ fetch_limit = min(max(result_limit * 2, result_limit),
_MAX_SOURCE_ROWS)
+
+ items: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ evidence: list[dict[str, Any]] = []
+ possibly_truncated = False
+
+ async def read_source(
+ source: str,
+ statement: str,
+ normalizer: Any,
+ *,
+ database_context: str | None = None,
+ ) -> None:
+ nonlocal possibly_truncated
+ try:
+ rows = await self._execute(
+ statement,
+ max_rows=fetch_limit,
+ database_context=database_context,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(f"{source} evidence is unavailable
({exc.reason_code}).")
+ evidence.append(
+ {
+ "source": source,
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ return
+ normalized_items = [
+ item
+ for row in rows
+ if (item := normalizer(row)) is not None
+ ]
+ items.extend(normalized_items)
+ possibly_truncated = possibly_truncated or len(rows) >= fetch_limit
+ evidence.append(
+ {
+ "source": source,
+ "success": True,
+ "rows_observed": len(rows),
+ }
+ )
+
+ quoted_database = quote_identifier(database_name, "database name")
+ if "batch_load" in selected_types:
+ await read_source(
+ "SHOW LOAD",
+ f"SHOW LOAD FROM {quoted_database} LIMIT {fetch_limit}",
+ lambda row: _batch_load_item(row, database_name),
+ )
+ if "stream_load" in selected_types:
+ await read_source(
+ "SHOW STREAM LOAD",
+ f"SHOW STREAM LOAD FROM {quoted_database} LIMIT {fetch_limit}",
+ _stream_load_item,
+ )
+ if "routine_load" in selected_types:
+ await read_source(
+ "SHOW ALL ROUTINE LOAD",
+ "SHOW ALL ROUTINE LOAD",
+ _routine_load_item,
+ database_context=database_name,
+ )
+ if {"insert_job", "continuous_load"} & selected_types:
+ try:
+ rows = await self._execute(
+ 'SELECT * FROM jobs("type"="insert") '
+ "ORDER BY CreateTime DESC LIMIT %s",
+ params=(fetch_limit,),
+ max_rows=fetch_limit,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ "Insert/continuous job evidence is unavailable "
+ f"({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": 'jobs("type"="insert")',
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ for row in rows:
+ item = _insert_job_item(row, database_name)
+ if item["job_type"] in selected_types:
+ items.append(item)
+ possibly_truncated = possibly_truncated or len(rows) >=
fetch_limit
+ evidence.append(
+ {
+ "source": 'jobs("type"="insert")',
+ "success": True,
+ "rows_observed": len(rows),
+ }
+ )
+
+ if not any(item.get("success") for item in evidence):
+ raise PipelineRuntimeFailure(
+ "Doris ingestion evidence is unavailable.",
+ reason_code="PIPELINE_INGESTION_EVIDENCE_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+
+ filtered: list[dict[str, Any]] = []
+ omitted_unscoped = 0
+ for item in items:
+ if selected_states and _state(item) not in selected_states:
+ continue
+ if table_name is not None:
+ observed_table = item.get("table")
+ if not isinstance(observed_table, str):
+ omitted_unscoped += 1
+ continue
+ if observed_table.casefold() != table_name.casefold():
+ continue
+ observed_database = item.get("database")
+ if isinstance(observed_database, str) and (
+ observed_database.casefold() != database_name.casefold()
+ ):
+ continue
+ filtered.append(item)
+ if omitted_unscoped:
+ warnings.append(
+ "Ingestion rows without an explicit target table were omitted "
+ "from the table-scoped result."
+ )
+
+ filtered.sort(key=_job_sort_key, reverse=True)
+ truncated = possibly_truncated or len(filtered) > result_limit
+ return _collection(
+ filtered[:result_limit],
+ sources=[item["source"] for item in evidence if
item.get("success")],
+ warnings=warnings,
+ truncated=truncated,
+ metadata={
+ "database": database_name,
+ "requested_job_types": sorted(selected_types),
+ "evidence": evidence,
+ },
+ )
+
+ async def diagnose_ingestion(
+ self,
+ *,
+ job_id: str | None,
+ database: str | None,
+ table: str | None,
+ window_minutes: int | None,
+ include_dependencies: bool,
+ ) -> dict[str, Any]:
+ normalized_job_id = _optional_text(job_id, "job identifier")
+ table_name = _optional_identifier(table, "table name")
+ if normalized_job_id is None and (database is None or table_name is
None):
+ raise PipelineRuntimeFailure(
+ "Provide job_id or both database and table.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ lookback = _bounded_int(
+ window_minutes,
+ default=_DEFAULT_DIAGNOSIS_WINDOW_MINUTES,
+ minimum=1,
+ maximum=43_200,
+ )
+ database_name = await self._resolve_database(database)
+ status_result = await self.get_ingestion_status(
+ job_types=None,
+ database=database_name,
+ table=table_name,
+ states=None,
+ limit=_MAX_LIMIT,
+ )
+ jobs = list(status_result["data"]["items"])
+ if normalized_job_id is not None:
+ jobs = [
+ item
+ for item in jobs
+ if str(item.get("job_id", "")).casefold()
+ == normalized_job_id.casefold()
+ or str(item.get("label", "")).casefold()
+ == normalized_job_id.casefold()
+ or str(item.get("name", "")).casefold()
+ == normalized_job_id.casefold()
+ ]
+
+ warnings = list(status_result["warnings"])
+ evidence: list[dict[str, Any]] = [
+ {
+ "source": "ingestion_status",
+ "matched_jobs": len(jobs),
+ }
+ ]
+ freshness: dict[str, Any] | None = None
+ dependencies: dict[str, Any] | None = None
+ if table_name is not None:
+ try:
+ freshness_result = await self.monitor_data_freshness(
+ database=database_name,
+ table=table_name,
+ threshold_seconds=None,
+ time_column=None,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Freshness evidence is unavailable ({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "freshness",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ freshness = freshness_result["data"]
+ warnings.extend(freshness_result["warnings"])
+ evidence.extend(freshness_result["evidence"])
+ if include_dependencies:
+ try:
+ dependency_result = await self.analyze_data_dependencies(
+ catalog=None,
+ database=database_name,
+ object_name=table_name,
+ direction="upstream",
+ depth=2,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Dependency evidence is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "dependencies",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ dependencies = dependency_result["data"]
+ warnings.extend(dependency_result["warnings"])
+ evidence.extend(dependency_result["evidence"])
+
+ audit_events, audit_warning = await self._recent_ingestion_audit(
+ job_id=normalized_job_id,
+ database=database_name,
+ table=table_name,
+ window_minutes=lookback,
+ )
+ if audit_warning:
+ warnings.append(audit_warning)
+ evidence.append(
+ {
+ "source": "internal.__internal_schema.audit_log",
+ "rows_observed": len(audit_events),
+ "success": audit_warning is None,
+ }
+ )
+
+ findings = _ingestion_findings(jobs, freshness)
+ if not jobs:
+ warnings.append("No ingestion job matched the requested scope.")
+ data: dict[str, Any] = {
+ "scope": {
+ "job_id": normalized_job_id,
+ "database": database_name,
+ "table": table_name,
+ "window_minutes": lookback,
+ },
+ "summary": {
+ "matched_jobs": len(jobs),
+ "failed_jobs": sum(_state(item) in _ERROR_STATES for item in
jobs),
+ "running_jobs": sum(_state(item) in _RUNNING_STATES for item
in jobs),
+ "completed_jobs": sum(
+ _state(item) in _SUCCESS_STATES for item in jobs
+ ),
+ "finding_count": len(findings),
+ },
+ "jobs": jobs,
+ "freshness": freshness,
+ "dependencies": dependencies,
+ "recent_audit_events": audit_events,
+ "findings": findings,
+ }
+ return _result(
+ data,
+ source="deterministic_ingestion_diagnosis",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={"invented_evidence": False},
+ )
+
+ async def get_materialized_view_status(
+ self,
+ *,
+ database: str | None,
+ view: str | None,
+ states: Sequence[str] | None,
+ include_refresh_history: bool,
+ ) -> dict[str, Any]:
+ database_name = await self._resolve_database(database)
+ view_name = _optional_identifier(view, "materialized view name")
+ selected_states = _normalized_filter(states)
+ database_property = _doris_property_string(database_name)
+ source_warnings: list[str] = []
+ successful_sources: list[str] = []
+ source_evidence: list[dict[str, Any]] = []
+
+ async def optional_rows(
+ source: str,
+ statement: str,
+ *,
+ max_rows: int = _MAX_SOURCE_ROWS,
+ params: tuple[Any, ...] | None = None,
+ ) -> list[dict[str, Any]]:
+ try:
+ rows = await self._execute(
+ statement,
+ params=params,
+ max_rows=max_rows,
+ )
+ except PipelineRuntimeFailure as exc:
+ source_warnings.append(
+ f"{source} evidence is unavailable ({exc.reason_code})."
+ )
+ source_evidence.append(
+ {
+ "source": source,
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ return []
+ successful_sources.append(source)
+ source_evidence.append(
+ {"source": source, "success": True, "rows_observed": len(rows)}
+ )
+ return rows
+
+ infos = await optional_rows(
+ 'mv_infos("database"=...)',
+ # SQL sink audit: database_name is accepted only after
+ # validate_identifier and encoded as a Doris TVF property before
+ # DorisConnection.execute.
+ (
+ "SELECT Id, Name, JobName, State, SchemaChangeDetail, " #
nosec B608
+ "RefreshState, RefreshInfo, SyncWithBaseTables "
+ f'FROM mv_infos("database"={database_property}) LIMIT '
+ "500"
+ ),
+ )
+ jobs = await optional_rows(
+ 'jobs("type"="mv")',
+ (
+ "SELECT Id, Name, MvId, MvName, MvDatabaseName, ExecuteType, "
+ "RecurringStrategy, Status, CreateTime "
+ 'FROM jobs("type"="mv") '
+ "WHERE MvDatabaseName = %s LIMIT 500"
+ ),
+ params=(database_name,),
+ )
+ tasks: list[dict[str, Any]] = []
+ if include_refresh_history:
+ tasks = await optional_rows(
+ 'tasks("type"="mv")',
+ (
+ "SELECT TaskId, JobId, JobName, MvId, MvName, "
+ "MvDatabaseName, Status, ErrorMsg, CreateTime, StartTime, "
+ "FinishTime, DurationMs, RefreshMode, "
+ "NeedRefreshPartitions, CompletedPartitions, Progress, "
+ "LastQueryId "
+ 'FROM tasks("type"="mv") '
+ "WHERE MvDatabaseName = %s ORDER BY CreateTime DESC LIMIT "
+ "500"
+ ),
+ params=(database_name,),
+ )
+
+ quoted_database = quote_identifier(database_name, "database name")
+ sync_rows = await optional_rows(
+ "SHOW ALTER TABLE MATERIALIZED VIEW",
+ f"SHOW ALTER TABLE MATERIALIZED VIEW FROM {quoted_database}",
+ )
+ if not successful_sources:
+ raise PipelineRuntimeFailure(
+ "Doris materialized-view evidence is unavailable.",
+ reason_code="PIPELINE_MV_EVIDENCE_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+
+ jobs_by_view = {
+ str(_value(row, "mv_name", "MvName", "name", "Name")).casefold():
row
+ for row in jobs
+ if _value(row, "mv_name", "MvName", "name", "Name") not in (None,
"")
+ }
+ tasks_by_view: dict[str, list[Mapping[str, Any]]] = {}
+ for row in tasks:
+ name = _value(row, "mv_name", "MvName")
+ if name not in (None, ""):
+ tasks_by_view.setdefault(str(name).casefold(), []).append(row)
+
+ items: list[dict[str, Any]] = []
+ seen_views: set[str] = set()
+ for row in infos:
+ normalized = _normalized_row(row)
+ name = str(normalized.get("name", ""))
+ if not name:
+ continue
+ seen_views.add(name.casefold())
+ job = jobs_by_view.get(name.casefold())
+ history = tasks_by_view.get(name.casefold(), [])
+ item = _async_mv_item(
+ normalized,
+ _normalized_row(job) if job is not None else None,
+ [_normalized_row(task) for task in history],
+ database_name=database_name,
+ include_history=include_refresh_history,
+ )
+ items.append(item)
+ for row in jobs:
+ normalized = _normalized_row(row)
+ name = str(normalized.get("mv_name") or normalized.get("name") or
"")
+ if not name or name.casefold() in seen_views:
+ continue
+ items.append(
+ _async_mv_item(
+ None,
+ normalized,
+ [
+ _normalized_row(task)
+ for task in tasks_by_view.get(name.casefold(), [])
+ ],
+ database_name=database_name,
+ include_history=include_refresh_history,
+ )
+ )
+ items.extend(_sync_mv_item(row, database_name) for row in sync_rows)
+
+ filtered = [
+ item
+ for item in items
+ if (
+ view_name is None
+ or str(item.get("view", "")).casefold() == view_name.casefold()
+ )
+ and (
+ not selected_states
+ or str(item.get("state", "")).casefold() in selected_states
+ or str(item.get("refresh_state", "")).casefold()
+ in selected_states
+ )
+ ]
+ filtered.sort(
+ key=lambda item: (
+ str(item.get("database", "")),
+ str(item.get("view", "")),
+ str(item.get("kind", "")),
+ )
+ )
+ truncated = len(filtered) > _MAX_LIMIT or any(
+ item.get("rows_observed", 0) >= _MAX_SOURCE_ROWS
+ for item in source_evidence
+ )
+ return _collection(
+ filtered[:_MAX_LIMIT],
+ sources=successful_sources,
+ warnings=source_warnings,
+ truncated=truncated,
+ metadata={
+ "database": database_name,
+ "include_refresh_history": include_refresh_history,
+ "evidence": source_evidence,
+ },
+ )
+
+ async def monitor_data_freshness(
+ self,
+ *,
+ database: str,
+ table: str,
+ threshold_seconds: int | None,
+ time_column: str | None,
+ ) -> dict[str, Any]:
+ database_name = _required_identifier(database, "database name")
+ table_name = _required_identifier(table, "table name")
+ threshold = _bounded_int(
+ threshold_seconds,
+ default=_DEFAULT_FRESHNESS_THRESHOLD_SECONDS,
+ minimum=1,
+ maximum=31_536_000,
+ )
+ column_name = _optional_identifier(time_column, "time column")
+ table_reference = build_table_reference(table_name, database_name)
+ warnings: list[str] = []
+ evidence: list[dict[str, Any]] = []
+ observed_at: Any | None = None
+ lag_seconds: int | None = None
+ method = "unknown"
+ partitions_observed = 0
+
+ if column_name is not None:
+ quoted_column = quote_identifier(column_name, "time column")
+ try:
+ # SQL sink audit: column_name and the table components pass
+ # quote_identifier/build_table_reference before
+ # DorisConnection.execute.
+ rows = await self._execute(
+ (
+ f"SELECT MAX({quoted_column}) AS observed_at, " #
nosec B608
+ f"TIMESTAMPDIFF(SECOND, MAX({quoted_column}), NOW()) "
+ f"AS lag_seconds FROM {table_reference}"
+ ),
+ max_rows=1,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Event-time evidence is unavailable ({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": f"MAX({column_name})",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ if rows:
+ normalized = _normalized_row(rows[0])
+ observed_at = normalized.get("observed_at")
+ lag_seconds = _as_int(normalized.get("lag_seconds"))
+ if observed_at is not None:
+ method = "event_time_column"
+ evidence.append(
+ {
+ "source": f"MAX({column_name})",
+ "success": observed_at is not None,
+ "evidence_quality": "recorded_event_time",
+ }
+ )
+
+ if observed_at is None:
+ try:
+ partitions = await self._execute(
+ f"SHOW PARTITIONS FROM {table_reference} LIMIT
{_MAX_PARTITIONS}",
+ max_rows=_MAX_PARTITIONS,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Visible-version evidence is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "SHOW PARTITIONS",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ timestamps = [
+ value
+ for row in partitions
+ if (
+ value := _value(
+ row,
+ "VisibleVersionTime",
+ "visible_version_time",
+ )
+ )
+ not in (None, "", "N/A")
+ ]
+ partitions_observed = len(partitions)
+ if timestamps:
+ observed_at = max(timestamps, key=lambda value: str(value))
+ lag_seconds = _seconds_since(observed_at)
+ method = "partition_visible_version"
+ if len(partitions) >= _MAX_PARTITIONS:
+ warnings.append(
+ "Partition evidence reached the server-side row limit."
+ )
+ evidence.append(
+ {
+ "source": "SHOW PARTITIONS.VisibleVersionTime",
+ "success": observed_at is not None,
+ "rows_observed": len(partitions),
+ "evidence_quality": "recorded_visible_version",
+ }
+ )
+
+ if observed_at is None:
+ pattern = f"%{_escape_like(table_name)}%"
+ try:
+ rows = await self._execute(
+ (
+ "SELECT MAX(`time`) AS observed_at "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE stmt LIKE %s AND "
+ "(UPPER(stmt) LIKE '%INSERT%' OR "
+ "UPPER(stmt) LIKE '%LOAD%')"
+ ),
+ params=(pattern,),
+ max_rows=1,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Audit freshness fallback is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "audit_log",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ if rows:
+ observed_at = _value(rows[0], "observed_at")
+ if observed_at is not None:
+ lag_seconds = _seconds_since(observed_at)
+ method = "audit_statement_time"
+ warnings.append(
+ "Freshness is inferred from an audit statement
timestamp, "
+ "not a committed data version."
+ )
+ evidence.append(
+ {
+ "source": "internal.__internal_schema.audit_log",
+ "success": observed_at is not None,
+ "evidence_quality": "inferred",
+ }
+ )
+
+ if observed_at is None or lag_seconds is None:
+ status = "unknown"
+ warnings.append("No recorded freshness evidence was found.")
+ elif lag_seconds <= threshold:
+ status = "fresh"
+ else:
+ status = "stale"
+
+ return _result(
+ {
+ "database": database_name,
+ "table": table_name,
+ "time_column": column_name,
+ "threshold_seconds": threshold,
+ "status": status,
+ "observed_at": _json_value(observed_at),
+ "lag_seconds": lag_seconds,
+ "method": method,
+ "partitions_observed": partitions_observed,
+ },
+ source=method,
+ warnings=warnings,
+ evidence=evidence,
+ metadata={"invented_evidence": False},
+ )
+
+ async def analyze_data_dependencies(
+ self,
+ *,
+ catalog: str | None,
+ database: str | None,
+ object_name: str,
+ direction: str | None,
+ depth: int | None,
+ ) -> dict[str, Any]:
+ catalog_name = _optional_identifier(catalog, "catalog name") or
"internal"
+ database_name = await self._resolve_database(database)
+ target_object = _required_identifier(object_name, "object name")
+ traversal_direction = direction or "both"
+ if traversal_direction not in {"upstream", "downstream", "both"}:
+ raise PipelineRuntimeFailure(
+ "Dependency direction must be upstream, downstream, or both.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ max_depth = _bounded_int(
+ depth,
+ default=3,
+ minimum=1,
+ maximum=_MAX_DEPENDENCY_DEPTH,
+ )
+ target = _canonical_object(
+ target_object,
+ default_catalog=catalog_name,
+ default_database=database_name,
+ )
+ warnings: list[str] = []
+ evidence: list[dict[str, Any]] = []
+ edges: list[dict[str, Any]] = []
+
+ if catalog_name == "internal":
+ database_property = _doris_property_string(database_name)
+ try:
+ # SQL sink audit: database_name is accepted only after
+ # validate_identifier and encoded as a Doris TVF property
+ # before DorisConnection.execute.
+ mv_rows = await self._execute(
+ (
+ "SELECT Name, QuerySql " # nosec B608
+ f'FROM mv_infos("database"={database_property}) '
+ "LIMIT 500"
+ ),
+ max_rows=_MAX_DEPENDENCY_EDGES,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Materialized-view dependency evidence is unavailable "
+ f"({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "mv_infos",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ edges.extend(
+ _declared_edges(
+ mv_rows,
+ target_key="Name",
+ sql_key="QuerySql",
+ default_catalog=catalog_name,
+ default_database=database_name,
+ evidence_source="mv_infos.QuerySql",
+ kind="materialized_view",
+ )
+ )
+ evidence.append(
+ {
+ "source": "mv_infos.QuerySql",
+ "success": True,
+ "rows_observed": len(mv_rows),
+ }
+ )
+
+ try:
+ view_rows = await self._execute(
+ (
+ "SELECT TABLE_NAME, VIEW_DEFINITION "
+ "FROM information_schema.views "
+ "WHERE TABLE_SCHEMA = %s "
+ "LIMIT 500"
+ ),
+ params=(database_name,),
+ max_rows=_MAX_DEPENDENCY_EDGES,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"View dependency evidence is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "information_schema.views",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ edges.extend(
+ _declared_edges(
+ view_rows,
+ target_key="TABLE_NAME",
+ sql_key="VIEW_DEFINITION",
+ default_catalog=catalog_name,
+ default_database=database_name,
+ evidence_source="information_schema.views",
+ kind="view",
+ )
+ )
+ evidence.append(
+ {
+ "source": "information_schema.views",
+ "success": True,
+ "rows_observed": len(view_rows),
+ }
+ )
+
+ try:
+ audit_rows = await self._execute(
+ (
+ "SELECT stmt, `time` AS observed_at "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 30 DAY) "
+ "AND (UPPER(stmt) LIKE '%INSERT%' OR "
+ "UPPER(stmt) LIKE '%CREATE%VIEW%') "
+ "ORDER BY `time` DESC "
+ "LIMIT 500"
+ ),
+ max_rows=_MAX_SOURCE_ROWS,
+ )
+ except PipelineRuntimeFailure as exc:
+ warnings.append(
+ f"Audit dependency evidence is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "source": "audit_log",
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ else:
+ edges.extend(
+ _audit_edges(
+ audit_rows,
+ default_catalog=catalog_name,
+ default_database=database_name,
+ )
+ )
+ evidence.append(
+ {
+ "source": "internal.__internal_schema.audit_log",
+ "success": True,
+ "rows_observed": len(audit_rows),
+ "evidence_quality": "inferred",
+ }
+ )
+
+ if not any(item.get("success") for item in evidence):
+ raise PipelineRuntimeFailure(
+ "Doris dependency evidence is unavailable.",
+ reason_code="PIPELINE_DEPENDENCY_EVIDENCE_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ merged_edges = _merge_edges(edges)[:_MAX_DEPENDENCY_EDGES]
+ graph = _traverse_dependencies(
+ target,
+ merged_edges,
+ direction=traversal_direction,
+ max_depth=max_depth,
+ )
+ if not graph["edges"]:
+ warnings.append("No recorded dependency evidence matched the
object.")
+ return _result(
+ {
+ "target": target,
+ "direction": traversal_direction,
+ "max_depth": max_depth,
+ **graph,
+ },
+ source="recorded_dependency_graph",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={
+ "invented_evidence": False,
+ "total_recorded_edges": len(merged_edges),
+ },
+ )
+
+ async def _recent_ingestion_audit(
+ self,
+ *,
+ job_id: str | None,
+ database: str,
+ table: str | None,
+ window_minutes: int,
+ ) -> tuple[list[dict[str, Any]], str | None]:
+ marker = table or job_id
+ if marker is None:
+ return [], None
+ pattern = f"%{_escape_like(marker)}%"
+ try:
+ rows = await self._execute(
+ (
+ "SELECT `time` AS observed_at, query_id, stmt "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s MINUTE) "
+ "AND stmt LIKE %s ORDER BY `time` DESC LIMIT 50"
+ ),
+ params=(window_minutes, pattern),
+ max_rows=50,
+ )
+ except PipelineRuntimeFailure as exc:
+ return [], f"Recent audit evidence is unavailable
({exc.reason_code})."
+ events: list[dict[str, Any]] = []
+ for row in rows:
+ statement = str(_value(row, "stmt") or "")
+ events.append(
+ {
+ "observed_at": _json_value(_value(row, "observed_at")),
+ "query_id": _value(row, "query_id"),
+ "operation": _statement_operation(statement),
+ "referenced_objects": _extract_sources(
+ statement,
+ default_catalog="internal",
+ default_database=database,
+ )[:20],
+ }
+ )
+ return events, None
+
+ async def _resolve_database(self, database: str | None) -> str:
+ if database is not None:
+ return _required_identifier(database, "database name")
+ try:
+ rows = await self._execute(
+ "SELECT DATABASE() AS database_name",
+ max_rows=1,
+ )
+ except PipelineRuntimeFailure as exc:
+ raise PipelineRuntimeFailure(
+ "A database is required for this Pipeline operation.",
+ reason_code="PIPELINE_DATABASE_REQUIRED",
+ status_code=400,
+ ) from exc
+ value = _value(rows[0], "database_name") if rows else None
+ if not isinstance(value, str) or not value:
+ raise PipelineRuntimeFailure(
+ "A database is required for this Pipeline operation.",
+ reason_code="PIPELINE_DATABASE_REQUIRED",
+ status_code=400,
+ )
+ return _required_identifier(value, "database name")
+
+ async def _execute(
+ self,
+ sql: str,
+ *,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ max_rows: int,
+ database_context: str | None = None,
+ ) -> list[dict[str, Any]]:
+ auth_context = get_current_auth_context()
+ session_id = f"{self._session_prefix}:{uuid.uuid4().hex[:8]}"
+ try:
+ async with (
+
self._connection_manager.get_connection_context_for_auth_context(
+ session_id,
+ auth_context,
+ ) as connection
+ ):
+ if database_context is not None:
+ safe_database = quote_identifier(
+ database_context,
+ "database name",
+ )
+ await connection.execute(
+ f"USE {safe_database}",
+ auth_context=auth_context,
+ mask_result=False,
+ max_rows=1,
+ max_bytes=1024,
+ )
+ result = await connection.execute(
+ sql,
+ params=params,
+ auth_context=auth_context,
+ mask_result=False,
+ max_rows=max_rows,
+ max_bytes=_MAX_BYTES,
+ )
+ except Exception as exc:
+ raise _classify_failure(exc) from exc
+ return [
+ dict(row)
+ for row in (result.data or ())
+ if isinstance(row, Mapping)
+ ][:max_rows]
+
+
+def _result(
+ data: Mapping[str, Any],
+ *,
+ source: str,
+ warnings: Sequence[str] = (),
+ metadata: Mapping[str, Any] | None = None,
+ evidence: Sequence[Mapping[str, Any]] | None = None,
+) -> dict[str, Any]:
+ unique_warnings = list(dict.fromkeys(str(item) for item in warnings))
+ response: dict[str, Any] = {
+ "status": "partial" if unique_warnings else "success",
+ "data": redact_sensitive_data(dict(data)),
+ "warnings": unique_warnings,
+ "metadata": {
+ "source": source,
+ **dict(metadata or {}),
+ },
+ }
+ if evidence is not None:
+ response["evidence"] = [
+ redact_sensitive_data(dict(item)) for item in evidence
+ ]
+ return response
+
+
+def _collection(
+ items: Sequence[Mapping[str, Any]],
+ *,
+ sources: Sequence[str],
+ warnings: Sequence[str] = (),
+ truncated: bool,
+ metadata: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+ return _result(
+ {
+ "items": [dict(item) for item in items],
+ "next_cursor": None,
+ "truncated": truncated,
+ },
+ source=",".join(dict.fromkeys(sources)) or "pipeline_metadata",
+ warnings=warnings,
+ metadata={
+ "returned_items": len(items),
+ **dict(metadata or {}),
+ },
+ )
+
+
+def _job_types(values: Sequence[str] | None) -> frozenset[str]:
+ if not values:
+ return _JOB_TYPES
+ normalized = frozenset(str(value).casefold() for value in values)
+ unsupported = normalized - _JOB_TYPES
+ if unsupported:
+ raise PipelineRuntimeFailure(
+ "Unsupported ingestion job type.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ return normalized
+
+
+def _normalized_filter(values: Sequence[str] | None) -> frozenset[str]:
+ return frozenset(str(value).strip().casefold() for value in values or ())
+
+
+def _bounded_int(
+ value: int | None,
+ *,
+ default: int,
+ minimum: int,
+ maximum: int,
+) -> int:
+ candidate = default if value is None else value
+ if isinstance(candidate, bool) or not isinstance(candidate, int):
+ raise PipelineRuntimeFailure(
+ "Pipeline numeric arguments must be integers.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ if candidate < minimum or candidate > maximum:
+ raise PipelineRuntimeFailure(
+ "Pipeline numeric argument is outside the supported range.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ return candidate
+
+
+def _required_identifier(value: str, label: str) -> str:
+ if not isinstance(value, str) or not value.strip():
+ raise PipelineRuntimeFailure(
+ f"{label.capitalize()} is required.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ try:
+ return validate_identifier(value.strip(), label)
+ except SQLSecurityError as exc:
+ raise PipelineRuntimeFailure(
+ f"Invalid {label}.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ ) from exc
+
+
+def _optional_identifier(value: str | None, label: str) -> str | None:
+ return None if value is None else _required_identifier(value, label)
+
+
+def _optional_text(value: str | None, label: str) -> str | None:
+ if value is None:
+ return None
+ if not isinstance(value, str) or not value.strip() or len(value) > 256:
+ raise PipelineRuntimeFailure(
+ f"Invalid {label}.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ return value.strip()
+
+
+def _normalized_row(row: Mapping[str, Any] | None) -> dict[str, Any]:
+ if row is None:
+ return {}
+ return {_snake_case(str(key)): value for key, value in row.items()}
+
+
+def _snake_case(value: str) -> str:
+ value = re.sub(r"(?<!^)(?=[A-Z])", "_", value)
+ return re.sub(r"[^a-z0-9]+", "_", value.casefold()).strip("_")
+
+
+def _value(row: Mapping[str, Any], *names: str) -> Any | None:
+ normalized = _normalized_row(row)
+ for name in names:
+ key = _snake_case(name)
+ if key in normalized:
+ return normalized[key]
+ return None
+
+
+def _batch_load_item(
+ row: Mapping[str, Any],
+ database_name: str,
+) -> dict[str, Any] | None:
+ item = _normalized_row(row)
+ job_id = item.get("job_id")
+ if job_id in (None, ""):
+ return None
+ target = _find_named_value(
+ (item.get("job_details"), item.get("task_info")),
+ {"table", "table_name", "tablename", "tbl"},
+ )
+ return {
+ "job_type": "batch_load",
+ "job_id": str(job_id),
+ "label": item.get("label"),
+ "database": database_name,
+ "table": target,
+ "state": _normalized_state(item.get("state")),
+ "progress": item.get("progress"),
+ "load_type": item.get("type"),
+ "create_time": _json_value(item.get("create_time")),
+ "start_time": _json_value(
+ item.get("load_start_time") or item.get("etl_start_time")
+ ),
+ "finish_time": _json_value(
+ item.get("load_finish_time") or item.get("etl_finish_time")
+ ),
+ "transaction_id": item.get("transaction_id"),
+ "error": item.get("first_error_msg") or item.get("error_msg"),
+ "comment": item.get("comment"),
+ }
+
+
+def _stream_load_item(row: Mapping[str, Any]) -> dict[str, Any] | None:
+ item = _normalized_row(row)
+ label = item.get("label")
+ if label in (None, ""):
+ return None
+ return {
+ "job_type": "stream_load",
+ "job_id": str(label),
+ "label": label,
+ "database": item.get("db"),
+ "table": item.get("table"),
+ "state": _normalized_state(item.get("status")),
+ "total_rows": _as_int(item.get("total_rows")),
+ "loaded_rows": _as_int(item.get("loaded_rows")),
+ "filtered_rows": _as_int(item.get("filtered_rows")),
+ "unselected_rows": _as_int(item.get("unselected_rows")),
+ "load_bytes": _as_int(item.get("load_bytes")),
+ "start_time": _json_value(item.get("start_time")),
+ "finish_time": _json_value(item.get("finish_time")),
+ "error": item.get("first_error_msg") or item.get("message"),
+ "comment": item.get("comment"),
+ }
+
+
+def _routine_load_item(row: Mapping[str, Any]) -> dict[str, Any] | None:
+ item = _normalized_row(row)
+ job_id = item.get("id")
+ if job_id in (None, ""):
+ return None
+ return {
+ "job_type": "routine_load",
+ "job_id": str(job_id),
+ "name": item.get("name"),
+ "database": item.get("db_name"),
+ "table": item.get("table_name"),
+ "state": _normalized_state(item.get("state")),
+ "data_source_type": item.get("data_source_type"),
+ "current_task_count": _as_int(item.get("current_task_num")),
+ "lag": _safe_json(item.get("lag")),
+ "progress": _safe_json(item.get("progress")),
+ "statistics": _safe_json(item.get("statistic")),
+ "create_time": _json_value(item.get("create_time")),
+ "pause_time": _json_value(item.get("pause_time")),
+ "finish_time": _json_value(item.get("end_time")),
+ "error": item.get("reason_of_state_changed") or item.get("other_msg"),
+ "compute_group": item.get("compute_group"),
+ "comment": item.get("comment"),
+ }
+
+
+def _insert_job_item(
+ row: Mapping[str, Any],
+ database_name: str,
+) -> dict[str, Any]:
+ item = _normalized_row(row)
+ statement = str(item.get("execute_sql") or "")
+ target = _extract_write_target(
+ statement,
+ default_catalog="internal",
+ default_database=database_name,
+ )
+ current_offset = item.get("current_offset")
+ end_offset = item.get("end_offset")
+ runtime_message = str(item.get("job_runtime_msg") or "")
+ continuous = any(
+ value not in (None, "", "null", "NULL")
+ for value in (current_offset, end_offset)
+ ) or "stream" in runtime_message.casefold()
+ target_parts = target.split(".") if target else []
+ return {
+ "job_type": "continuous_load" if continuous else "insert_job",
+ "job_id": str(item.get("id") or item.get("name") or ""),
+ "name": item.get("name"),
+ "database": target_parts[-2] if len(target_parts) >= 2 else None,
+ "table": target_parts[-1] if target_parts else None,
+ "state": _normalized_state(item.get("status")),
+ "execute_type": item.get("execute_type"),
+ "recurring_strategy": item.get("recurring_strategy"),
+ "create_time": _json_value(item.get("create_time")),
+ "succeeded_task_count": _as_int(item.get("succeed_task_count")),
+ "failed_task_count": _as_int(item.get("failed_task_count")),
+ "canceled_task_count": _as_int(item.get("canceled_task_count")),
+ "current_offset": _safe_json(current_offset),
+ "end_offset": _safe_json(end_offset),
+ "load_statistics": _safe_json(item.get("load_statistic")),
+ "error": item.get("error_msg"),
+ "comment": item.get("comment"),
+ }
+
+
+def _async_mv_item(
+ info: Mapping[str, Any] | None,
+ job: Mapping[str, Any] | None,
+ tasks: Sequence[Mapping[str, Any]],
+ *,
+ database_name: str,
+ include_history: bool,
+) -> dict[str, Any]:
+ view = (
+ (info or {}).get("name")
+ or (job or {}).get("mv_name")
+ or (job or {}).get("name")
+ )
+ history = [_mv_task_item(task) for task in tasks[:20]]
+ item: dict[str, Any] = {
+ "kind": "asynchronous",
+ "database": database_name,
+ "view": view,
+ "state": _normalized_state(
+ (info or {}).get("state") or (job or {}).get("status")
+ ),
+ "refresh_state": _normalized_state((info or {}).get("refresh_state")),
+ "sync_with_base_tables": _as_bool(
+ (info or {}).get("sync_with_base_tables")
+ ),
+ "refresh_info": (info or {}).get("refresh_info"),
+ "schema_change_detail": (info or {}).get("schema_change_detail"),
+ "job": {
+ "job_id": (job or {}).get("id"),
+ "job_name": (info or {}).get("job_name")
+ or (job or {}).get("name"),
+ "status": _normalized_state((job or {}).get("status")),
+ "execute_type": (job or {}).get("execute_type"),
+ "recurring_strategy": (job or {}).get("recurring_strategy"),
+ "create_time": _json_value((job or {}).get("create_time")),
+ },
+ "latest_refresh": history[0] if history else None,
+ }
+ if include_history:
+ item["refresh_history"] = history
+ return item
+
+
+def _mv_task_item(task: Mapping[str, Any]) -> dict[str, Any]:
+ return {
+ "task_id": task.get("task_id"),
+ "status": _normalized_state(task.get("status")),
+ "error": task.get("error_msg"),
+ "create_time": _json_value(task.get("create_time")),
+ "start_time": _json_value(task.get("start_time")),
+ "finish_time": _json_value(task.get("finish_time")),
+ "duration_ms": _as_int(task.get("duration_ms")),
+ "refresh_mode": task.get("refresh_mode"),
+ "progress": task.get("progress"),
+ "needed_partitions": _safe_json(task.get("need_refresh_partitions")),
+ "completed_partitions": _safe_json(task.get("completed_partitions")),
+ "last_query_id": task.get("last_query_id"),
+ }
+
+
+def _sync_mv_item(
+ row: Mapping[str, Any],
+ database_name: str,
+) -> dict[str, Any]:
+ item = _normalized_row(row)
+ return {
+ "kind": "synchronous_build",
+ "database": database_name,
+ "view": item.get("rollup_index_name") or item.get("index_name"),
+ "base_table": item.get("table_name"),
+ "job_id": item.get("job_id"),
+ "state": _normalized_state(item.get("state")),
+ "progress": item.get("progress"),
+ "create_time": _json_value(item.get("create_time")),
+ "finish_time": _json_value(item.get("finish_time")),
+ "error": item.get("msg"),
+ }
+
+
+def _ingestion_findings(
+ jobs: Sequence[Mapping[str, Any]],
+ freshness: Mapping[str, Any] | None,
+) -> list[dict[str, Any]]:
+ findings: list[dict[str, Any]] = []
+ for item in jobs:
+ state = _state(item)
+ if state in _ERROR_STATES:
+ findings.append(
+ {
+ "severity": "high",
+ "code": "INGESTION_JOB_NOT_HEALTHY",
+ "job_id": item.get("job_id"),
+ "state": state,
+ "message": "The recorded ingestion job is not healthy.",
+ }
+ )
+ lag = _numeric_total(item.get("lag"))
+ if lag is not None and lag > 0:
+ findings.append(
+ {
+ "severity": "medium",
+ "code": "ROUTINE_LOAD_LAG_RECORDED",
+ "job_id": item.get("job_id"),
+ "lag_total": lag,
+ "message": "Routine Load reports a positive source lag.",
+ }
+ )
+ failed_tasks = _as_int(item.get("failed_task_count"))
+ if failed_tasks is not None and failed_tasks > 0:
+ findings.append(
+ {
+ "severity": "medium",
+ "code": "INGESTION_TASK_FAILURES_RECORDED",
+ "job_id": item.get("job_id"),
+ "failed_task_count": failed_tasks,
+ "message": "The ingestion job records failed tasks.",
+ }
+ )
+ if freshness and freshness.get("status") == "stale":
+ findings.append(
+ {
+ "severity": "medium",
+ "code": "TARGET_TABLE_STALE",
+ "lag_seconds": freshness.get("lag_seconds"),
+ "message": "The target table exceeds the freshness threshold.",
+ }
+ )
+ return findings
+
+
+def _declared_edges(
+ rows: Sequence[Mapping[str, Any]],
+ *,
+ target_key: str,
+ sql_key: str,
+ default_catalog: str,
+ default_database: str,
+ evidence_source: str,
+ kind: str,
+) -> list[dict[str, Any]]:
+ edges: list[dict[str, Any]] = []
+ for row in rows:
+ target_value = _value(row, target_key)
+ statement = _value(row, sql_key)
+ if not isinstance(target_value, str) or not isinstance(statement, str):
+ continue
+ target = _canonical_object(
+ target_value,
+ default_catalog=default_catalog,
+ default_database=default_database,
+ )
+ for source in _extract_sources(
+ statement,
+ default_catalog=default_catalog,
+ default_database=default_database,
+ ):
+ if source != target:
+ edges.append(
+ {
+ "source": source,
+ "target": target,
+ "kind": kind,
+ "confidence": 1.0,
+ "evidence_source": evidence_source,
+ "observed_at": None,
+ }
+ )
+ return edges
+
+
+def _audit_edges(
+ rows: Sequence[Mapping[str, Any]],
+ *,
+ default_catalog: str,
+ default_database: str,
+) -> list[dict[str, Any]]:
+ edges: list[dict[str, Any]] = []
+ for row in rows:
+ statement = _value(row, "stmt")
+ if not isinstance(statement, str):
+ continue
+ target = _extract_write_target(
+ statement,
+ default_catalog=default_catalog,
+ default_database=default_database,
+ )
+ if target is None:
+ continue
+ for source in _extract_sources(
+ statement,
+ default_catalog=default_catalog,
+ default_database=default_database,
+ ):
+ if source != target:
+ edges.append(
+ {
+ "source": source,
+ "target": target,
+ "kind": "audit_sql_inference",
+ "confidence": 0.7,
+ "evidence_source":
"internal.__internal_schema.audit_log",
+ "observed_at": _json_value(_value(row, "observed_at")),
+ }
+ )
+ return edges
+
+
+def _merge_edges(edges: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
+ merged: dict[tuple[str, str], dict[str, Any]] = {}
+ for edge in edges:
+ source = str(edge.get("source", ""))
+ target = str(edge.get("target", ""))
+ if not source or not target:
+ continue
+ key = (source, target)
+ current = merged.get(key)
+ if current is None:
+ merged[key] = {
+ "source": source,
+ "target": target,
+ "kind": edge.get("kind"),
+ "confidence": edge.get("confidence"),
+ "evidence_sources": [edge.get("evidence_source")],
+ "observed_at": edge.get("observed_at"),
+ }
+ continue
+ current["confidence"] = max(
+ float(current.get("confidence") or 0),
+ float(edge.get("confidence") or 0),
+ )
+ current["evidence_sources"] = list(
+ dict.fromkeys(
+ [
+ *current["evidence_sources"],
+ edge.get("evidence_source"),
+ ]
+ )
+ )
+ if edge.get("observed_at") is not None:
+ current["observed_at"] = max(
+ str(current.get("observed_at") or ""),
+ str(edge["observed_at"]),
+ )
+ return sorted(merged.values(), key=lambda item: (item["source"],
item["target"]))
+
+
+def _traverse_dependencies(
+ target: str,
+ edges: Sequence[Mapping[str, Any]],
+ *,
+ direction: str,
+ max_depth: int,
+) -> dict[str, Any]:
+ adjacency_up: dict[str, list[Mapping[str, Any]]] = {}
+ adjacency_down: dict[str, list[Mapping[str, Any]]] = {}
+ for edge in edges:
+ source = str(edge["source"])
+ destination = str(edge["target"])
+ adjacency_up.setdefault(destination, []).append(edge)
+ adjacency_down.setdefault(source, []).append(edge)
+
+ selected_edges: dict[tuple[str, str], Mapping[str, Any]] = {}
+ paths: list[dict[str, Any]] = []
+ directions = (
+ ("upstream", adjacency_up)
+ if direction == "upstream"
+ else (
+ ("downstream", adjacency_down)
+ if direction == "downstream"
+ else None
+ )
+ )
+ direction_pairs = (
+ (directions,)
+ if directions is not None
+ else (("upstream", adjacency_up), ("downstream", adjacency_down))
+ )
+ for traversal_name, adjacency in direction_pairs:
+ queue: deque[tuple[str, list[str], int]] = deque([(target, [target],
0)])
+ visited: set[tuple[str, int]] = {(target, 0)}
+ while queue:
+ current, path, current_depth = queue.popleft()
+ if current_depth >= max_depth:
+ continue
+ for edge in adjacency.get(current, ()):
+ source = str(edge["source"])
+ destination = str(edge["target"])
+ next_node = (
+ source if traversal_name == "upstream" else destination
+ )
+ selected_edges[(source, destination)] = edge
+ next_path = [*path, next_node]
+ paths.append(
+ {
+ "direction": traversal_name,
+ "depth": current_depth + 1,
+ "nodes": next_path,
+ }
+ )
+ marker = (next_node, current_depth + 1)
+ if marker not in visited:
+ visited.add(marker)
+ queue.append((next_node, next_path, current_depth + 1))
+
+ edge_items = sorted(
+ (dict(edge) for edge in selected_edges.values()),
+ key=lambda item: (item["source"], item["target"]),
+ )
+ nodes = sorted(
+ {
+ target,
+ *(
+ str(value)
+ for edge in edge_items
+ for value in (edge["source"], edge["target"])
+ ),
+ }
+ )
+ paths.sort(key=lambda item: (item["direction"], item["depth"],
item["nodes"]))
+ return {"nodes": nodes, "edges": edge_items, "paths": paths}
+
+
+def _extract_sources(
+ sql: str,
+ *,
+ default_catalog: str,
+ default_database: str,
+) -> list[str]:
+ cleaned = _sql_structure(sql)
+ cte_names = {
+ _strip_identifier(match.group(1)).casefold()
+ for match in _CTE_NAME.finditer(cleaned)
+ }
+ sources = []
+ for match in _SOURCE_REFERENCE.finditer(cleaned):
+ raw = match.group(1)
+ if _strip_identifier(raw).casefold() in cte_names:
+ continue
+ try:
+ source = _canonical_object(
+ raw,
+ default_catalog=default_catalog,
+ default_database=default_database,
+ )
+ except PipelineRuntimeFailure:
+ continue
+ sources.append(source)
+ return list(dict.fromkeys(sources))
+
+
+def _extract_write_target(
+ sql: str,
+ *,
+ default_catalog: str,
+ default_database: str,
+) -> str | None:
+ cleaned = _sql_structure(sql)
+ match = _WRITE_TARGET.search(cleaned)
+ if match is None:
+ return None
+ try:
+ return _canonical_object(
+ match.group(1),
+ default_catalog=default_catalog,
+ default_database=default_database,
+ )
+ except PipelineRuntimeFailure:
+ return None
+
+
+def _sql_structure(sql: str) -> str:
+ statements = sqlparse.parse(sql)
+ if not statements:
+ return ""
+ parts: list[str] = []
+ for token in statements[0].flatten():
+ token_type = token.ttype
+ if token_type in sql_tokens.Comment or token_type in
sql_tokens.Literal.String:
+ parts.append(" ")
+ else:
+ parts.append(token.value)
+ return "".join(parts)
+
+
+def _canonical_object(
+ value: str,
+ *,
+ default_catalog: str,
+ default_database: str,
+) -> str:
+ parts = [
+ _strip_identifier(part)
+ for part in re.split(r"\s*\.\s*", value.strip())
+ if part.strip()
+ ]
+ if len(parts) == 1:
+ parts = [default_catalog, default_database, parts[0]]
+ elif len(parts) == 2:
+ parts = [default_catalog, *parts]
+ elif len(parts) != 3:
+ raise PipelineRuntimeFailure(
+ "Invalid object reference.",
+ reason_code="PIPELINE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+ validated = [
+ _required_identifier(part, "object identifier")
+ for part in parts
+ ]
+ return ".".join(validated)
+
+
+def _strip_identifier(value: str) -> str:
+ return value.strip().strip("`").strip('"')
+
+
+def _statement_operation(statement: str) -> str:
+ cleaned = _sql_structure(statement).lstrip()
+ match = re.match(r"([A-Za-z]+)", cleaned)
+ return match.group(1).upper() if match else "UNKNOWN"
+
+
+def _find_named_value(
+ values: Sequence[Any],
+ keys: set[str],
+) -> str | None:
+ for value in values:
+ parsed = _safe_json(value)
+ if isinstance(parsed, Mapping):
+ for key, candidate in parsed.items():
+ if _snake_case(str(key)) in keys and isinstance(candidate,
str):
+ try:
+ return validate_identifier(candidate, "table name")
+ except SQLSecurityError:
+ continue
+ return None
+
+
+def _safe_json(value: Any) -> Any:
+ if not isinstance(value, str):
+ return _json_value(value)
+ stripped = value.strip()
+ if not stripped:
+ return None
+ try:
+ parsed = json.loads(stripped)
+ except (TypeError, ValueError):
+ return stripped[:512]
+ return _json_value(parsed)
+
+
+def _json_value(value: Any) -> Any:
+ if isinstance(value, datetime):
+ return value.isoformat()
+ if isinstance(value, Mapping):
+ return {str(key): _json_value(item) for key, item in value.items()}
+ if isinstance(value, Sequence) and not isinstance(value, str | bytes):
+ return [_json_value(item) for item in value]
+ if isinstance(value, bytes):
+ return value.decode("utf-8", errors="replace")
+ return value
+
+
+def _normalized_state(value: Any) -> str | None:
+ if value in (None, ""):
+ return None
+ return str(value).strip().casefold()
+
+
+def _state(item: Mapping[str, Any]) -> str:
+ return str(item.get("state") or "").casefold()
+
+
+def _job_sort_key(item: Mapping[str, Any]) -> str:
+ for key in ("finish_time", "start_time", "create_time"):
+ value = item.get(key)
+ if value not in (None, ""):
+ return str(value)
+ return ""
+
+
+def _as_int(value: Any) -> int | None:
+ if value in (None, "") or isinstance(value, bool):
+ return None
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ return None
+
+
+def _as_bool(value: Any) -> bool | None:
+ if isinstance(value, bool):
+ return value
+ if isinstance(value, int):
+ return value != 0
+ if isinstance(value, str):
+ if value.casefold() in {"1", "true", "yes"}:
+ return True
+ if value.casefold() in {"0", "false", "no"}:
+ return False
+ return None
+
+
+def _numeric_total(value: Any) -> float | None:
+ if isinstance(value, int | float) and not isinstance(value, bool):
+ return float(value)
+ if isinstance(value, Mapping):
+ numbers = [
+ float(item)
+ for item in value.values()
+ if isinstance(item, int | float) and not isinstance(item, bool)
+ ]
+ return sum(numbers) if numbers else None
+ return None
+
+
+def _seconds_since(value: Any) -> int | None:
+ if isinstance(value, datetime):
+ observed = value
+ elif isinstance(value, str):
+ try:
+ observed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ except ValueError:
+ return None
+ else:
+ return None
+ now = datetime.now(UTC) if observed.tzinfo is not None else datetime.now()
+ return max(0, int((now - observed).total_seconds()))
+
+
+def _escape_like(value: str) -> str:
+ return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+def _doris_property_string(value: str) -> str:
+ return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+def _classify_failure(exc: Exception) -> PipelineRuntimeFailure:
+ if isinstance(exc, PipelineRuntimeFailure):
+ return exc
+ numeric_code = next(
+ (value for value in getattr(exc, "args", ()) if isinstance(value,
int)),
+ None,
+ )
+ message = str(exc).casefold()
+ if numeric_code in {1044, 1045, 1142, 1227} or any(
+ marker in message
+ for marker in ("access denied", "permission denied", "privilege")
+ ):
+ return PipelineRuntimeFailure(
+ "Doris denied access to Pipeline metadata.",
+ reason_code="PIPELINE_PERMISSION_DENIED",
+ status_code=403,
+ )
+ if numeric_code in {1064, 1109, 1146} or any(
+ marker in message
+ for marker in (
+ "doesn't exist",
+ "does not exist",
+ "not supported",
+ "unsupported",
+ "unknown table",
+ "no viable alternative",
+ )
+ ):
+ return PipelineRuntimeFailure(
+ "The requested Doris Pipeline evidence source is unsupported.",
+ reason_code="PIPELINE_SOURCE_UNSUPPORTED",
+ status_code=501,
+ )
+ if isinstance(exc, TimeoutError | ConnectionError | OSError):
+ return PipelineRuntimeFailure(
+ "Doris Pipeline evidence is temporarily unavailable.",
+ reason_code="PIPELINE_BACKEND_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ return PipelineRuntimeFailure(
+ "Doris Pipeline metadata execution failed.",
+ reason_code="PIPELINE_EXECUTION_FAILED",
+ status_code=502,
+ )
+
+
+__all__ = ["DorisPipelineRuntime", "PipelineRuntimeFailure"]
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index eb9f40f..cfbfa50 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -84,6 +84,13 @@ CLUSTER_CHILD_NAMES = (
"analyze_resource_growth",
"get_runtime_capabilities",
)
+PIPELINE_CHILD_NAMES = (
+ "get_ingestion_status",
+ "diagnose_ingestion",
+ "get_materialized_view_status",
+ "monitor_data_freshness",
+ "analyze_data_dependencies",
+)
@dataclass(frozen=True)
@@ -1188,6 +1195,121 @@ async def
test_real_doris_hierarchical_cluster_domain_is_read_only_and_live(
assert compaction["data"]["native_tracker"] is False
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_hierarchical_pipeline_domain_is_read_only_and_live(
+ transport: str,
+ doris_sandbox: DorisSandbox,
+) -> None:
+ environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.settings.user,
+ password=doris_sandbox.settings.password,
+ )
+ environment["MCP_TOOL_EXPOSURE_MODE"] = "hierarchical"
+ with doris_sandbox.admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"INSERT INTO {doris_sandbox.qualified_table} VALUES (1, %s)",
+ (doris_sandbox.marker,),
+ )
+
+ async with _transport_client(
+ transport,
+ environment,
+ read_timeout_seconds=60,
+ ) as client:
+ pipeline_result = await client.call_tool("doris_pipeline", {})
+ assert pipeline_result.is_error is False
+ assert isinstance(pipeline_result.structured_content, dict)
+ manifest = pipeline_result.structured_content
+ assert manifest["mode"] == "manifest"
+ assert manifest["domain"] == "doris_pipeline"
+ children = {child["name"]: child for child in manifest["children"]}
+ assert tuple(children) == PIPELINE_CHILD_NAMES
+ assert all(child["availability"]["callable"] for child in
children.values())
+ manifest_version = manifest["manifest_version"]
+
+ ingestion = await _call_domain_child(
+ client,
+ domain="doris_pipeline",
+ child_tool="get_ingestion_status",
+ arguments={
+ "database": doris_sandbox.settings.database,
+ "table": doris_sandbox.table,
+ "limit": 20,
+ },
+ manifest_version=manifest_version,
+ )
+ assert ingestion["metadata"]["database"] == (
+ doris_sandbox.settings.database
+ )
+ assert ingestion["data"]["truncated"] is False
+
+ diagnosis = await _call_domain_child(
+ client,
+ domain="doris_pipeline",
+ child_tool="diagnose_ingestion",
+ arguments={
+ "database": doris_sandbox.settings.database,
+ "table": doris_sandbox.table,
+ "window_minutes": 60,
+ "include_dependencies": False,
+ },
+ manifest_version=manifest_version,
+ )
+ assert diagnosis["data"]["summary"]["matched_jobs"] >= 0
+ assert diagnosis["metadata"]["invented_evidence"] is False
+
+ freshness = await _call_domain_child(
+ client,
+ domain="doris_pipeline",
+ child_tool="monitor_data_freshness",
+ arguments={
+ "database": doris_sandbox.settings.database,
+ "table": doris_sandbox.table,
+ "threshold_seconds": 3600,
+ },
+ manifest_version=manifest_version,
+ )
+ assert freshness["data"]["status"] in {"fresh", "stale"}
+ assert freshness["data"]["method"] == "partition_visible_version"
+ assert freshness["metadata"]["invented_evidence"] is False
+
+ materialized_views = await _call_domain_child(
+ client,
+ domain="doris_pipeline",
+ child_tool="get_materialized_view_status",
+ arguments={
+ "database": doris_sandbox.settings.database,
+ "include_refresh_history": True,
+ },
+ manifest_version=manifest_version,
+ )
+ assert isinstance(materialized_views["data"]["items"], list)
+
+ dependencies = await _call_domain_child(
+ client,
+ domain="doris_pipeline",
+ child_tool="analyze_data_dependencies",
+ arguments={
+ "database": doris_sandbox.settings.database,
+ "object": doris_sandbox.table,
+ "direction": "both",
+ "depth": 2,
+ },
+ manifest_version=manifest_version,
+ )
+ assert dependencies["metadata"]["invented_evidence"] is False
+ assert "unknown_source" not in str(dependencies)
+
+ with doris_sandbox.admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"SELECT COUNT(*) AS row_count "
+ f"FROM {doris_sandbox.qualified_table}"
+ )
+ row = cursor.fetchone()
+ assert row == (1,)
+
+
@pytest.mark.skipif(
os.getenv("DORIS_REAL_HTTP_INTEGRATION") != "1",
reason="set DORIS_REAL_HTTP_INTEGRATION=1 with independent FE/BE HTTP
endpoints",
diff --git a/test/tools/test_capability_detector.py
b/test/tools/test_capability_detector.py
index 6124029..025aa54 100644
--- a/test/tools/test_capability_detector.py
+++ b/test/tools/test_capability_detector.py
@@ -125,6 +125,7 @@ class _ProbeConnectionManager:
fingerprint="route-a",
)
self.context_error: Exception | None = None
+ self.context_sessions: list[str] = []
def get_route_identity(self, _auth_context: Any) -> DorisRouteIdentity:
return self.route
@@ -138,11 +139,12 @@ class _ProbeConnectionManager:
@asynccontextmanager
async def get_connection_context_for_auth_context(
self,
- _session_id: str,
+ session_id: str,
_auth_context: Any,
):
if self.context_error is not None:
raise self.context_error
+ self.context_sessions.append(session_id)
yield self.connection
@@ -204,6 +206,84 @@ async def
test_detector_builds_version_vector_and_extends_domains_lazily() -> No
assert connection.statements.count("SELECT @@version_comment;") == 1
[email protected]
+async def test_pipeline_probes_isolate_an_unsupported_source_connection() ->
None:
+ base_connection = _ProbeConnection()
+
+ class _PoisoningConnection(_ProbeConnection):
+ def __init__(self, *, fail_load: bool) -> None:
+ super().__init__()
+ self.fail_load = fail_load
+ self.poisoned = False
+
+ async def execute(
+ self,
+ sql: str,
+ *_args: Any,
+ **_kwargs: Any,
+ ) -> SimpleNamespace:
+ if self.poisoned:
+ raise ConnectionError("connection was poisoned")
+ if self.fail_load and sql == "SHOW LOAD LIMIT 1":
+ self.poisoned = True
+ raise RuntimeError(1064, "unsupported statement")
+ return await super().execute(sql, *_args, **_kwargs)
+
+ class _IsolatingManager(_ProbeConnectionManager):
+ def __init__(self) -> None:
+ super().__init__(base_connection)
+ self.domain_connections: list[_PoisoningConnection] = []
+
+ @asynccontextmanager
+ async def get_connection_context_for_auth_context(
+ self,
+ session_id: str,
+ _auth_context: Any,
+ ):
+ self.context_sessions.append(session_id)
+ if not session_id.startswith("capability-domain:doris_pipeline:"):
+ yield self.connection
+ return
+ connection = _PoisoningConnection(
+ fail_load=":0:" in session_id,
+ )
+ self.domain_connections.append(connection)
+ yield connection
+
+ manager = _IsolatingManager()
+ detector = DorisCapabilityDetector(manager) # type: ignore[arg-type]
+ base = await detector.detect_base(
+ None,
+ capability_generation=1,
+ provider_generation="provider.a",
+ )
+
+ pipeline = await detector.detect_domain(base, "doris_pipeline", None)
+
+ assert (
+ pipeline.probe("batch_load_metadata_readable").status
+ is CapabilityProbeStatus.UNSUPPORTED
+ )
+ assert (
+ pipeline.probe("stream_load_metadata_readable").status
+ is CapabilityProbeStatus.SUPPORTED
+ )
+ assert (
+ pipeline.probe("stream_broker_routine_load_readable").status
+ is CapabilityProbeStatus.SUPPORTED
+ )
+ assert (
+ pipeline.probe("ingestion_status_readable").status
+ is CapabilityProbeStatus.SUPPORTED
+ )
+ assert (
+ pipeline.probe("materialized_view_tasks_readable").status
+ is CapabilityProbeStatus.SUPPORTED
+ )
+ assert len(manager.domain_connections) == 7
+ assert len({id(connection) for connection in manager.domain_connections})
== 7
+
+
@pytest.mark.asyncio
async def test_detector_marks_permission_failure_unknown_not_unsupported() ->
None:
connection = _ProbeConnection()
diff --git a/test/tools/test_domain_dispatcher.py
b/test/tools/test_domain_dispatcher.py
index 83b7738..ac094e2 100644
--- a/test/tools/test_domain_dispatcher.py
+++ b/test/tools/test_domain_dispatcher.py
@@ -523,13 +523,17 @@ async def
test_manifest_version_unavailable_and_unbound_fail_closed() -> None:
},
)
)
- unbound_manager = _manager("doris_pipeline.get_ingestion_status")
+ unbound_manager = _manager("doris_search.search_data")
unbound = json.loads(
await unbound_manager.call_tool(
- "doris_pipeline",
+ "doris_search",
{
- "child_tool": "get_ingestion_status",
- "arguments": {},
+ "child_tool": "search_data",
+ "arguments": {
+ "database": "analytics",
+ "table": "events",
+ "mode": "text",
+ },
},
)
)
@@ -543,6 +547,51 @@ async def
test_manifest_version_unavailable_and_unbound_fail_closed() -> None:
assert unbound["error"]["details"]["reason_code"] == "HANDLER_NOT_BOUND"
[email protected]
+async def test_formal_pipeline_handler_overrides_legacy_migration_binding() ->
None:
+ manager = _manager("doris_pipeline.monitor_data_freshness")
+ formal_result = {
+ "status": "success",
+ "data": {
+ "database": "analytics",
+ "table": "orders",
+ "status": "fresh",
+ },
+ "warnings": [],
+ "metadata": {"source": "partition_visible_version"},
+ "evidence": [],
+ }
+ manager.pipeline_runtime.monitor_data_freshness = AsyncMock(
+ return_value=formal_result
+ )
+ manager.data_governance_tools.monitor_data_freshness = AsyncMock(
+ side_effect=AssertionError("legacy Pipeline handler must not run")
+ )
+
+ response = json.loads(
+ await manager.call_tool(
+ "doris_pipeline",
+ {
+ "child_tool": "monitor_data_freshness",
+ "arguments": {
+ "database": "analytics",
+ "table": "orders",
+ },
+ },
+ )
+ )
+
+ assert response["mode"] == "result"
+ assert response["data"] == formal_result
+ manager.pipeline_runtime.monitor_data_freshness.assert_awaited_once_with(
+ database="analytics",
+ table="orders",
+ threshold_seconds=None,
+ time_column=None,
+ )
+ manager.data_governance_tools.monitor_data_freshness.assert_not_awaited()
+
+
@pytest.mark.asyncio
async def test_runtime_provider_down_is_rechecked_before_execution() -> None:
provider = SelectiveAvailabilityProvider({"doris_query.execute_query"})
diff --git a/test/utils/test_pipeline_runtime.py
b/test/utils/test_pipeline_runtime.py
new file mode 100644
index 0000000..d378a8d
--- /dev/null
+++ b/test/utils/test_pipeline_runtime.py
@@ -0,0 +1,348 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Production evidence contracts for the formal Pipeline runtime."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Mapping
+from contextlib import asynccontextmanager
+from datetime import datetime
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from doris_mcp_server.utils.pipeline_runtime import (
+ DorisPipelineRuntime,
+ PipelineRuntimeFailure,
+)
+
+
+class _ConnectionManager:
+ def __init__(
+ self,
+ *,
+ rows: dict[str, list[dict[str, Any]]] | None = None,
+ failures: dict[str, Exception] | None = None,
+ ) -> None:
+ self.rows = rows or {}
+ self.failures = failures or {}
+ self.calls: list[str] = []
+ self.params: list[Mapping[str, Any] | tuple[Any, ...] | None] = []
+ self.context_count = 0
+
+ @asynccontextmanager
+ async def get_connection_context_for_auth_context(
+ self,
+ _session_id: str,
+ _auth_context: Any,
+ ) -> AsyncIterator[Any]:
+ self.context_count += 1
+ manager = self
+
+ class _Connection:
+ async def execute(
+ self,
+ sql: str,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ **_kwargs: Any,
+ ) -> SimpleNamespace:
+ manager.calls.append(sql)
+ manager.params.append(params)
+ if sql in manager.failures:
+ raise manager.failures[sql]
+ return SimpleNamespace(data=manager.rows.get(sql, []))
+
+ yield _Connection()
+
+
+def _runtime(
+ *,
+ rows: dict[str, list[dict[str, Any]]] | None = None,
+ failures: dict[str, Exception] | None = None,
+) -> tuple[DorisPipelineRuntime, _ConnectionManager]:
+ manager = _ConnectionManager(rows=rows, failures=failures)
+ return DorisPipelineRuntime(manager), manager # type: ignore[arg-type]
+
+
[email protected]
+async def
test_ingestion_status_normalizes_real_rows_and_omits_sensitive_fields() -> None:
+ batch_sql = "SHOW LOAD FROM `analytics` LIMIT 20"
+ routine_sql = "SHOW ALL ROUTINE LOAD"
+ runtime, manager = _runtime(
+ rows={
+ batch_sql: [
+ {
+ "JobId": 101,
+ "Label": "orders_20260731",
+ "State": "FINISHED",
+ "Progress": "100%",
+ "JobDetails": '{"table_name":"orders"}',
+ "URL": "http://fe:8030/api/load?token=secret";,
+ "CreateTime": "2026-07-31 01:00:00",
+ "LoadFinishTime": "2026-07-31 01:01:00",
+ }
+ ],
+ routine_sql: [
+ {
+ "Id": 202,
+ "Name": "orders_kafka",
+ "DbName": "analytics",
+ "TableName": "orders",
+ "State": "RUNNING",
+ "DataSourceType": "KAFKA",
+ "CurrentTaskNum": 3,
+ "Lag": '{"0": 4}',
+ "Progress": '{"0": "17"}',
+ "ErrorLogUrls": "http://be:8040/error?password=secret";,
+ },
+ {
+ "Id": 203,
+ "Name": "other_db",
+ "DbName": "other",
+ "TableName": "orders",
+ "State": "RUNNING",
+ },
+ ],
+ }
+ )
+
+ result = await runtime.get_ingestion_status(
+ job_types=["batch_load", "routine_load"],
+ database="analytics",
+ table="orders",
+ states=None,
+ limit=10,
+ )
+
+ assert result["status"] == "success"
+ assert [item["job_type"] for item in result["data"]["items"]] == [
+ "batch_load",
+ "routine_load",
+ ]
+ assert result["data"]["items"][1]["lag"] == {"0": 4}
+ serialized = str(result)
+ assert "ErrorLogUrls" not in serialized
+ assert "http://fe:8030"; not in serialized
+ assert "password=secret" not in serialized
+ assert manager.calls == [
+ batch_sql,
+ "USE `analytics`",
+ routine_sql,
+ ]
+ assert manager.context_count == 2
+
+
[email protected]
+async def test_ingestion_status_fails_closed_without_any_evidence_source() ->
None:
+ sql = "SHOW LOAD FROM `analytics` LIMIT 20"
+ runtime, _ = _runtime(
+ failures={sql: RuntimeError(1146, "source does not exist")}
+ )
+
+ with pytest.raises(
+ PipelineRuntimeFailure,
+ match="ingestion evidence is unavailable",
+ ) as failure:
+ await runtime.get_ingestion_status(
+ job_types=["batch_load"],
+ database="analytics",
+ table=None,
+ states=None,
+ limit=10,
+ )
+
+ assert failure.value.reason_code ==
"PIPELINE_INGESTION_EVIDENCE_UNAVAILABLE"
+
+
[email protected]
+async def test_freshness_prefers_recorded_partition_visibility_time() -> None:
+ sql = "SHOW PARTITIONS FROM `analytics`.`orders` LIMIT 1000"
+ now = datetime.now()
+ runtime, _ = _runtime(
+ rows={
+ sql: [
+ {
+ "PartitionName": "p20260731",
+ "VisibleVersion": 12,
+ "VisibleVersionTime": now,
+ }
+ ]
+ }
+ )
+
+ result = await runtime.monitor_data_freshness(
+ database="analytics",
+ table="orders",
+ threshold_seconds=60,
+ time_column=None,
+ )
+
+ assert result["data"]["status"] == "fresh"
+ assert result["data"]["method"] == "partition_visible_version"
+ assert result["data"]["partitions_observed"] == 1
+ assert result["metadata"]["invented_evidence"] is False
+
+
[email protected]
+async def test_freshness_rejects_identifier_injection_before_execution() ->
None:
+ runtime, manager = _runtime()
+
+ with pytest.raises(PipelineRuntimeFailure) as failure:
+ await runtime.monitor_data_freshness(
+ database="analytics",
+ table="orders; DROP TABLE users",
+ threshold_seconds=60,
+ time_column=None,
+ )
+
+ assert failure.value.reason_code == "PIPELINE_ARGUMENT_INVALID"
+ assert manager.calls == []
+
+
[email protected]
+async def test_materialized_view_status_uses_recorded_jobs_and_refresh_tasks()
-> None:
+ info_sql = (
+ "SELECT Id, Name, JobName, State, SchemaChangeDetail, RefreshState, "
+ 'RefreshInfo, SyncWithBaseTables FROM mv_infos("database"="analytics")
'
+ "LIMIT 500"
+ )
+ job_sql = (
+ "SELECT Id, Name, MvId, MvName, MvDatabaseName, ExecuteType, "
+ 'RecurringStrategy, Status, CreateTime FROM jobs("type"="mv") '
+ "WHERE MvDatabaseName = %s LIMIT 500"
+ )
+ task_sql = (
+ "SELECT TaskId, JobId, JobName, MvId, MvName, MvDatabaseName, Status, "
+ "ErrorMsg, CreateTime, StartTime, FinishTime, DurationMs, RefreshMode,
"
+ "NeedRefreshPartitions, CompletedPartitions, Progress, LastQueryId "
+ 'FROM tasks("type"="mv") WHERE MvDatabaseName = %s '
+ "ORDER BY CreateTime DESC LIMIT 500"
+ )
+ sync_sql = "SHOW ALTER TABLE MATERIALIZED VIEW FROM `analytics`"
+ runtime, manager = _runtime(
+ rows={
+ info_sql: [
+ {
+ "Id": 1,
+ "Name": "mv_orders",
+ "JobName": "inner_mtmv_1",
+ "State": "NORMAL",
+ "RefreshState": "SUCCESS",
+ "SyncWithBaseTables": "true",
+ }
+ ],
+ job_sql: [
+ {
+ "Id": 11,
+ "Name": "inner_mtmv_1",
+ "MvName": "mv_orders",
+ "MvDatabaseName": "analytics",
+ "Status": "RUNNING",
+ }
+ ],
+ task_sql: [
+ {
+ "TaskId": 21,
+ "MvName": "mv_orders",
+ "Status": "SUCCESS",
+ "DurationMs": 45,
+ "LastQueryId": "query-1",
+ }
+ ],
+ sync_sql: [],
+ }
+ )
+
+ result = await runtime.get_materialized_view_status(
+ database="analytics",
+ view="mv_orders",
+ states=None,
+ include_refresh_history=True,
+ )
+
+ item = result["data"]["items"][0]
+ assert item["view"] == "mv_orders"
+ assert item["sync_with_base_tables"] is True
+ assert item["latest_refresh"]["task_id"] == 21
+ assert item["refresh_history"][0]["duration_ms"] == 45
+ assert manager.params[1] == ("analytics",)
+ assert manager.params[2] == ("analytics",)
+ assert "query_sql" not in str(result).casefold()
+
+
[email protected]
+async def
test_dependency_graph_uses_only_recorded_edges_without_synthetic_nodes() ->
None:
+ mv_sql = (
+ 'SELECT Name, QuerySql FROM mv_infos("database"="analytics") LIMIT 500'
+ )
+ view_sql = (
+ "SELECT TABLE_NAME, VIEW_DEFINITION FROM information_schema.views "
+ "WHERE TABLE_SCHEMA = %s LIMIT 500"
+ )
+ audit_sql = (
+ "SELECT stmt, `time` AS observed_at "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 30 DAY) "
+ "AND (UPPER(stmt) LIKE '%INSERT%' OR "
+ "UPPER(stmt) LIKE '%CREATE%VIEW%') "
+ "ORDER BY `time` DESC LIMIT 500"
+ )
+ runtime, _ = _runtime(
+ rows={
+ mv_sql: [
+ {
+ "Name": "daily_orders",
+ "QuerySql": "SELECT * FROM raw_orders",
+ }
+ ],
+ view_sql: [
+ {
+ "TABLE_NAME": "order_view",
+ "VIEW_DEFINITION": "SELECT * FROM analytics.daily_orders",
+ }
+ ],
+ audit_sql: [
+ {
+ "stmt": (
+ "INSERT INTO final_orders "
+ "SELECT * FROM analytics.order_view"
+ ),
+ "observed_at": "2026-07-31 02:00:00",
+ }
+ ],
+ }
+ )
+
+ result = await runtime.analyze_data_dependencies(
+ catalog="internal",
+ database="analytics",
+ object_name="final_orders",
+ direction="upstream",
+ depth=5,
+ )
+
+ assert result["data"]["nodes"] == [
+ "internal.analytics.daily_orders",
+ "internal.analytics.final_orders",
+ "internal.analytics.order_view",
+ "internal.analytics.raw_orders",
+ ]
+ assert len(result["data"]["edges"]) == 3
+ assert "unknown_source" not in str(result)
+ assert result["metadata"]["invented_evidence"] is False
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
