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 c873a85 feat: add Lakehouse domain runtime (#180)
c873a85 is described below
commit c873a85388b2591c926d37fe0d6dacfbc23a68fa
Author: Yijia Su <[email protected]>
AuthorDate: Fri Jul 31 20:34:53 2026 +0800
feat: add Lakehouse domain runtime (#180)
---
.env.example | 15 +
CHANGELOG.md | 11 +
README.md | 22 +
doris_mcp_server/main.py | 33 +
doris_mcp_server/tools/capability_detector.py | 78 ++
doris_mcp_server/tools/domain_dispatcher.py | 29 +
doris_mcp_server/tools/doris_feature_matrix.py | 21 +-
doris_mcp_server/tools/lakehouse_handlers.py | 83 ++
doris_mcp_server/tools/tools_manager.py | 3 +
doris_mcp_server/utils/config.py | 85 ++
doris_mcp_server/utils/lakehouse_runtime.py | 1722 ++++++++++++++++++++++++
test/integration/test_real_doris_transports.py | 173 +++
test/protocol/test_multiworker_config.py | 90 ++
test/tools/test_capability_detector.py | 58 +
test/tools/test_capability_registry.py | 133 ++
test/tools/test_domain_dispatcher.py | 43 +
test/tools/test_doris_feature_matrix.py | 45 +
test/tools/test_lakehouse_handlers.py | 163 +++
test/utils/test_lakehouse_runtime.py | 519 +++++++
19 files changed, 3317 insertions(+), 9 deletions(-)
diff --git a/.env.example b/.env.example
index 2694f5e..b090727 100644
--- a/.env.example
+++ b/.env.example
@@ -86,6 +86,21 @@ CAPABILITY_SNAPSHOT_TTL_SECONDS=300
CAPABILITY_PROBE_TIMEOUT_SECONDS=5
CAPABILITY_STALE_GRACE_SECONDS=900
+# Read-only Governance runtime bounds and optional native lineage store.
+GOVERNANCE_MAX_SAMPLE_RATIO=0.25
+GOVERNANCE_MAX_AUDIT_WINDOW_DAYS=30
+GOVERNANCE_MAX_LINEAGE_EDGES=500
+GOVERNANCE_LINEAGE_STORE_TABLE=
+GOVERNANCE_LINEAGE_RECENT_EVENT_MINUTES=1440
+
+# Read-only Lakehouse catalog, snapshot, partition, and Variant bounds.
+LAKEHOUSE_MAX_CATALOG_OBJECTS=50
+LAKEHOUSE_MAX_CATALOG_DATABASES=20
+LAKEHOUSE_MAX_SNAPSHOTS=50
+LAKEHOUSE_MAX_PARTITIONS=100
+LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS=20
+LAKEHOUSE_MAX_VARIANT_PATHS=200
+
# ===================================================================
# Security Configuration
# ===================================================================
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dac5f85..0b044b2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -62,6 +62,9 @@ under **Unreleased** until a new version is selected and
published.
patterns, audit events, user-defined functions, and authentication mappings.
- An explicit queryable lineage-provider contract and canonical Doris lineage
event-store schema for companion-plugin deployments.
+- A read-only Lakehouse domain with capability-gated external-catalog,
+ lakehouse-table, snapshot, partition, pushdown, and Variant-shape
+ inspection.
- Real Doris process tests covering Streamable HTTP and stdio.
### Changed
@@ -102,6 +105,9 @@ under **Unreleased** until a new version is selected and
published.
versions and live plugin, store, and audit evidence: Doris 4.0.6 and later
can use native companion-plugin events, while audit inference remains the
primary path before 4.0.6 and an explicit degraded fallback afterward.
+- Selected Lakehouse and Variant capability variants from observed Doris
+ component versions and live metadata probes, with 4.1 lifecycle and
+ advanced Variant facets reported separately from target-level evidence.
### Fixed
@@ -124,6 +130,11 @@ under **Unreleased** until a new version is selected and
published.
- Emitted lineage edges only from attributable native events or conservative
direct-column audit evidence, without placeholder sources or invented
numeric confidence scores.
+- Kept catalog property values, storage locations, raw plans, and sampled
+ Variant values out of model-facing output while bounding object, snapshot,
+ partition, path, and type-shape evidence.
+- Preserved configured Governance and Lakehouse runtime limits across the
+ multi-worker parent-to-worker environment handoff.
- 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/README.md b/README.md
index b7e9a79..51dd681 100644
--- a/README.md
+++ b/README.md
@@ -340,6 +340,28 @@ cp .env.example .env
probe phase (default: 5; range: 1-60)
* `CAPABILITY_STALE_GRACE_SECONDS`: Bounded interval in which a failed
refresh may reuse stale private evidence (default: 900; range: 0-86400)
+ * `GOVERNANCE_MAX_SAMPLE_RATIO`: Maximum live column-analysis sample
+ ratio (default: 0.25; range: greater than 0 through 1)
+ * `GOVERNANCE_MAX_AUDIT_WINDOW_DAYS`: Maximum audit-analysis window
+ (default: 30; range: 1-365)
+ * `GOVERNANCE_MAX_LINEAGE_EDGES`: Maximum returned lineage edges
+ (default: 500; range: 1-5000)
+ * `GOVERNANCE_LINEAGE_STORE_TABLE`: Optional canonical queryable
+ companion-plugin lineage event table
+ * `GOVERNANCE_LINEAGE_RECENT_EVENT_MINUTES`: Native lineage-provider
+ recency window (default: 1440; range: 1-525600)
+ * `LAKEHOUSE_MAX_CATALOG_OBJECTS`: Maximum sampled relations per
+ external-catalog inspection (default: 50; range: 1-500)
+ * `LAKEHOUSE_MAX_CATALOG_DATABASES`: Maximum sampled databases per
+ external catalog (default: 20; range: 1-100)
+ * `LAKEHOUSE_MAX_SNAPSHOTS`: Maximum returned lakehouse snapshots
+ (default: 50; range: 1-500)
+ * `LAKEHOUSE_MAX_PARTITIONS`: Maximum returned lakehouse partitions
+ (default: 100; range: 1-1000)
+ * `LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS`: Maximum rows inspected through
+ `VARIANT_TYPE` without returning values (default: 20; range: 1-500)
+ * `LAKEHOUSE_MAX_VARIANT_PATHS`: Maximum returned Variant type-shape
+ paths (default: 200; range: 1-2000)
* `MCP_STATE_HANDLE_SECRET`: Optional shared high-entropy key (at least
32 bytes) used to authenticate explicit cross-call state handles
* `MCP_STATE_HANDLE_TTL_SECONDS`: Lifetime of an explicit state handle
diff --git a/doris_mcp_server/main.py b/doris_mcp_server/main.py
index 9439ac4..a4b7a68 100644
--- a/doris_mcp_server/main.py
+++ b/doris_mcp_server/main.py
@@ -101,6 +101,39 @@ def _multiworker_environment(
"CAPABILITY_STALE_GRACE_SECONDS": str(
config.capability.stale_grace_seconds
),
+ "GOVERNANCE_MAX_SAMPLE_RATIO": str(
+ config.governance.max_sample_ratio
+ ),
+ "GOVERNANCE_MAX_AUDIT_WINDOW_DAYS": str(
+ config.governance.max_audit_window_days
+ ),
+ "GOVERNANCE_MAX_LINEAGE_EDGES": str(
+ config.governance.max_lineage_edges
+ ),
+ "GOVERNANCE_LINEAGE_STORE_TABLE": (
+ config.governance.lineage_store_table
+ ),
+ "GOVERNANCE_LINEAGE_RECENT_EVENT_MINUTES": str(
+ config.governance.lineage_recent_event_minutes
+ ),
+ "LAKEHOUSE_MAX_CATALOG_OBJECTS": str(
+ config.lakehouse.max_catalog_objects
+ ),
+ "LAKEHOUSE_MAX_CATALOG_DATABASES": str(
+ config.lakehouse.max_catalog_databases
+ ),
+ "LAKEHOUSE_MAX_SNAPSHOTS": str(
+ config.lakehouse.max_snapshots
+ ),
+ "LAKEHOUSE_MAX_PARTITIONS": str(
+ config.lakehouse.max_partitions
+ ),
+ "LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS": str(
+ config.lakehouse.max_variant_sample_rows
+ ),
+ "LAKEHOUSE_MAX_VARIANT_PATHS": str(
+ config.lakehouse.max_variant_paths
+ ),
"MCP_STATE_HANDLE_SECRET": config.mcp_state_handle_secret,
"MCP_STATE_HANDLE_TTL_SECONDS":
str(config.mcp_state_handle_ttl_seconds),
"SERVER_NAME": config.server_name,
diff --git a/doris_mcp_server/tools/capability_detector.py
b/doris_mcp_server/tools/capability_detector.py
index eafc49f..09fe0c2 100644
--- a/doris_mcp_server/tools/capability_detector.py
+++ b/doris_mcp_server/tools/capability_detector.py
@@ -347,6 +347,26 @@ _DOMAIN_PROBES: Mapping[str, tuple[tuple[str, tuple[str,
...]], ...]] = {
("lineage_plugin_config_readable",),
),
),
+ "doris_lakehouse": (
+ (
+ "SHOW CATALOGS",
+ ("external_catalog_metadata_readable",),
+ ),
+ (
+ (
+ "SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME "
+ "FROM information_schema.tables LIMIT 1"
+ ),
+ ("lakehouse_table_metadata_readable",),
+ ),
+ (
+ (
+ "SELECT COLUMN_NAME, DATA_TYPE "
+ "FROM information_schema.columns LIMIT 1"
+ ),
+ ("variant_column_type_readable",),
+ ),
+ ),
}
@@ -490,6 +510,10 @@ class DorisCapabilityDetector:
probes.update(
_combine_governance_evidence_probes(probes)
)
+ elif domain_name == "doris_lakehouse":
+ probes.update(
+ _combine_lakehouse_evidence_probes(probes)
+ )
completed_route = self.route_identity(auth_context)
if completed_route.fingerprint != base.route.fingerprint:
raise CapabilityRouteChangedError(
@@ -1905,6 +1929,60 @@ def _combine_governance_evidence_probes(
return derived
+def _combine_lakehouse_evidence_probes(
+ probes: Mapping[str, CapabilityProbeEvidence],
+) -> dict[str, CapabilityProbeEvidence]:
+ """Derive target-sensitive 4.1 facets from readable metadata surfaces."""
+ derived: dict[str, CapabilityProbeEvidence] = {}
+ table_metadata = probes.get("lakehouse_table_metadata_readable")
+ if table_metadata is not None:
+ for probe_id in (
+ "lakehouse_snapshot_features_readable",
+ "iceberg_deletion_vector",
+ "iceberg_row_lineage",
+ ):
+ derived[probe_id] = CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=(
+ CapabilityProbeStatus.DEGRADED
+ if table_metadata.status is CapabilityProbeStatus.SUPPORTED
+ else table_metadata.status
+ ),
+ reason_code=(
+ "TARGET_LAKEHOUSE_FORMAT_REQUIRES_CALL_TIME_VALIDATION"
+ if table_metadata.status is CapabilityProbeStatus.SUPPORTED
+ else table_metadata.reason_code
+ ),
+ evidence_sources=table_metadata.evidence_sources,
+ )
+ variant_metadata = probes.get("variant_column_type_readable")
+ if variant_metadata is not None:
+ for probe_id in (
+ "variant_advanced_properties_readable",
+ "variant_sparse_sharding",
+ "variant_sparse_cache",
+ "variant_doc_mode",
+ "storage_v3",
+ ):
+ derived[probe_id] = CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=(
+ CapabilityProbeStatus.DEGRADED
+ if variant_metadata.status
+ is CapabilityProbeStatus.SUPPORTED
+ else variant_metadata.status
+ ),
+ reason_code=(
+ "TARGET_VARIANT_PROPERTIES_REQUIRE_CALL_TIME_VALIDATION"
+ if variant_metadata.status
+ is CapabilityProbeStatus.SUPPORTED
+ else variant_metadata.reason_code
+ ),
+ evidence_sources=variant_metadata.evidence_sources,
+ )
+ return derived
+
+
def _combine_all_runtime_probes(
probe_id: str,
probes: Mapping[str, CapabilityProbeEvidence],
diff --git a/doris_mcp_server/tools/domain_dispatcher.py
b/doris_mcp_server/tools/domain_dispatcher.py
index 802498b..2144e16 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.governance_runtime import GovernanceRuntimeFailure
+from ..utils.lakehouse_runtime import LakehouseRuntimeFailure
from ..utils.logger import get_audit_logger, get_logger
from ..utils.pipeline_runtime import PipelineRuntimeFailure
from ..utils.query_runtime import QueryRuntimeFailure
@@ -736,6 +737,34 @@ class DomainDispatcher:
"status_code": exc.status_code,
},
)
+ except LakehouseRuntimeFailure as exc:
+ self._audit(feature_id, arguments, "error", started)
+ return self._error(
+ domain.name,
+ (
+ DomainErrorCode.CHILD_ARGUMENTS_INVALID
+ if exc.reason_code
+ in {
+ "LAKEHOUSE_ARGUMENT_INVALID",
+ "LAKEHOUSE_CATALOG_NOT_FOUND",
+ "LAKEHOUSE_CATALOG_NOT_EXTERNAL",
+ "LAKEHOUSE_TABLE_NOT_FOUND",
+ "LAKEHOUSE_TABLE_FORMAT_UNSUPPORTED",
+ "LAKEHOUSE_VARIANT_COLUMN_NOT_FOUND",
+ "LAKEHOUSE_COLUMN_NOT_VARIANT",
+ }
+ 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/doris_feature_matrix.py
b/doris_mcp_server/tools/doris_feature_matrix.py
index 1b09ac4..5480eff 100644
--- a/doris_mcp_server/tools/doris_feature_matrix.py
+++ b/doris_mcp_server/tools/doris_feature_matrix.py
@@ -1468,27 +1468,25 @@ FEATURE_DEFINITIONS = (
"doris_lakehouse",
"inspect_lakehouse_table",
A,
- _variant(
- "lakehouse_table_metadata",
- providers=("external_catalog_provider",),
- probes=("lakehouse_table_metadata_readable",),
- ),
_variant(
"lakehouse_lifecycle_4_1",
ranges=(">=4.1.0",),
+ providers=("external_catalog_provider",),
features=("iceberg_deletion_vector", "iceberg_row_lineage"),
probes=("lakehouse_snapshot_features_readable",),
+ callable_when_degraded=True,
sources=("DORIS_RELEASE_4_1_0",),
),
+ _variant(
+ "lakehouse_table_metadata",
+ providers=("external_catalog_provider",),
+ probes=("lakehouse_table_metadata_readable",),
+ ),
),
_feature(
"doris_lakehouse",
"inspect_variant_column",
A,
- _variant(
- "variant_type",
- probes=("variant_column_type_readable",),
- ),
_variant(
"variant_advanced_4_1",
ranges=(">=4.1.0",),
@@ -1499,8 +1497,13 @@ FEATURE_DEFINITIONS = (
"storage_v3",
),
probes=("variant_advanced_properties_readable",),
+ callable_when_degraded=True,
sources=("DORIS_RELEASE_4_1_0",),
),
+ _variant(
+ "variant_type",
+ probes=("variant_column_type_readable",),
+ ),
),
_feature(
"doris_semantic",
diff --git a/doris_mcp_server/tools/lakehouse_handlers.py
b/doris_mcp_server/tools/lakehouse_handlers.py
new file mode 100644
index 0000000..d6c7b4f
--- /dev/null
+++ b/doris_mcp_server/tools/lakehouse_handlers.py
@@ -0,0 +1,83 @@
+# 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 Lakehouse-domain handlers backed by one strict runtime."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, cast
+
+from ..utils.db import DorisConnectionManager
+from ..utils.lakehouse_runtime import DorisLakehouseRuntime
+
+
+class _LakehouseHandlerOwner(Protocol):
+ """State supplied by ``DorisToolsManager`` to the mixin."""
+
+ lakehouse_runtime: DorisLakehouseRuntime
+
+
+class LakehouseToolHandlersMixin:
+ """Route every Lakehouse child through the read-only runtime."""
+
+ def _initialize_lakehouse_handlers(
+ self: _LakehouseHandlerOwner,
+ connection_manager: DorisConnectionManager,
+ ) -> None:
+ self.lakehouse_runtime = DorisLakehouseRuntime(connection_manager)
+
+ async def _formal_doris_lakehouse_inspect_external_catalog_tool(
+ self: _LakehouseHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.lakehouse_runtime.inspect_external_catalog(
+ catalog=cast(str, arguments.get("catalog")),
+ include_objects=bool(arguments.get("include_objects", False)),
+ object_limit=cast(int | None, arguments.get("object_limit")),
+ )
+
+ async def _formal_doris_lakehouse_inspect_lakehouse_table_tool(
+ self: _LakehouseHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.lakehouse_runtime.inspect_lakehouse_table(
+ catalog=cast(str, arguments.get("catalog")),
+ database=cast(str, arguments.get("database")),
+ table=cast(str, arguments.get("table")),
+ include_snapshots=bool(
+ arguments.get("include_snapshots", False)
+ ),
+ include_partitions=bool(
+ arguments.get("include_partitions", False)
+ ),
+ )
+
+ async def _formal_doris_lakehouse_inspect_variant_column_tool(
+ self: _LakehouseHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.lakehouse_runtime.inspect_variant_column(
+ catalog=cast(str | None, arguments.get("catalog")),
+ database=cast(str, arguments.get("database")),
+ table=cast(str, arguments.get("table")),
+ column=cast(str, arguments.get("column")),
+ path=cast(str | None, arguments.get("path")),
+ sample_rows=cast(int | None, arguments.get("sample_rows")),
+ )
+
+
+__all__ = ["LakehouseToolHandlersMixin"]
diff --git a/doris_mcp_server/tools/tools_manager.py
b/doris_mcp_server/tools/tools_manager.py
index ec65c75..9cfeb3c 100644
--- a/doris_mcp_server/tools/tools_manager.py
+++ b/doris_mcp_server/tools/tools_manager.py
@@ -64,6 +64,7 @@ from .domain_manifest import (
)
from .doris_feature_matrix import DORIS_FEATURE_MATRIX
from .governance_handlers import GovernanceToolHandlersMixin
+from .lakehouse_handlers import LakehouseToolHandlersMixin
from .pipeline_handlers import PipelineToolHandlersMixin
from .query_handlers import QueryToolHandlersMixin
from .search_handlers import SearchToolHandlersMixin
@@ -80,6 +81,7 @@ class DorisToolsManager(
PipelineToolHandlersMixin,
SearchToolHandlersMixin,
GovernanceToolHandlersMixin,
+ LakehouseToolHandlersMixin,
DomainManifestManagerMixin,
):
"""Apache Doris Tools Manager"""
@@ -129,6 +131,7 @@ class DorisToolsManager(
self.query_runtime,
)
self._initialize_governance_handlers(connection_manager)
+ self._initialize_lakehouse_handlers(connection_manager)
self._capability_registry: CapabilityRegistry | None = None
if domain_availability_provider is None:
bound_handlers = BoundHandlerAvailabilityProvider(self)
diff --git a/doris_mcp_server/utils/config.py b/doris_mcp_server/utils/config.py
index 3caa4fb..45c035b 100644
--- a/doris_mcp_server/utils/config.py
+++ b/doris_mcp_server/utils/config.py
@@ -876,6 +876,18 @@ class GovernanceConfig:
lineage_recent_event_minutes: int = 1440
+@dataclass
+class LakehouseConfig:
+ """Read-only Lakehouse runtime collection and sampling limits."""
+
+ max_catalog_objects: int = 50
+ max_catalog_databases: int = 20
+ max_snapshots: int = 50
+ max_partitions: int = 100
+ max_variant_sample_rows: int = 20
+ max_variant_paths: int = 200
+
+
@dataclass
class DorisConfig:
"""Doris MCP Server complete configuration"""
@@ -916,6 +928,7 @@ class DorisConfig:
default_factory=CapabilityConfig
)
governance: GovernanceConfig = field(default_factory=GovernanceConfig)
+ lakehouse: LakehouseConfig = field(default_factory=LakehouseConfig)
# Custom configuration
custom_config: dict[str, Any] = field(default_factory=dict)
@@ -1614,6 +1627,36 @@ class DorisConfig:
"GOVERNANCE_LINEAGE_RECENT_EVENT_MINUTES",
config.governance.lineage_recent_event_minutes,
)
+ if "LAKEHOUSE_MAX_CATALOG_OBJECTS" in os.environ:
+ config.lakehouse.max_catalog_objects = _env_int(
+ "LAKEHOUSE_MAX_CATALOG_OBJECTS",
+ config.lakehouse.max_catalog_objects,
+ )
+ if "LAKEHOUSE_MAX_CATALOG_DATABASES" in os.environ:
+ config.lakehouse.max_catalog_databases = _env_int(
+ "LAKEHOUSE_MAX_CATALOG_DATABASES",
+ config.lakehouse.max_catalog_databases,
+ )
+ if "LAKEHOUSE_MAX_SNAPSHOTS" in os.environ:
+ config.lakehouse.max_snapshots = _env_int(
+ "LAKEHOUSE_MAX_SNAPSHOTS",
+ config.lakehouse.max_snapshots,
+ )
+ if "LAKEHOUSE_MAX_PARTITIONS" in os.environ:
+ config.lakehouse.max_partitions = _env_int(
+ "LAKEHOUSE_MAX_PARTITIONS",
+ config.lakehouse.max_partitions,
+ )
+ if "LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS" in os.environ:
+ config.lakehouse.max_variant_sample_rows = _env_int(
+ "LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS",
+ config.lakehouse.max_variant_sample_rows,
+ )
+ if "LAKEHOUSE_MAX_VARIANT_PATHS" in os.environ:
+ config.lakehouse.max_variant_paths = _env_int(
+ "LAKEHOUSE_MAX_VARIANT_PATHS",
+ config.lakehouse.max_variant_paths,
+ )
if "MCP_STATE_HANDLE_SECRET" in os.environ:
config.mcp_state_handle_secret = os.getenv(
"MCP_STATE_HANDLE_SECRET",
@@ -1732,6 +1775,12 @@ class DorisConfig:
if hasattr(config.governance, key):
setattr(config.governance, key, value)
+ if "lakehouse" in config_data:
+ lakehouse_config = config_data["lakehouse"]
+ for key, value in lakehouse_config.items():
+ if hasattr(config.lakehouse, key):
+ setattr(config.lakehouse, key, value)
+
# Custom configuration
config.custom_config = config_data.get("custom", {})
@@ -1776,6 +1825,18 @@ class DorisConfig:
self.governance.lineage_recent_event_minutes
),
},
+ "lakehouse": {
+ "max_catalog_objects": self.lakehouse.max_catalog_objects,
+ "max_catalog_databases": (
+ self.lakehouse.max_catalog_databases
+ ),
+ "max_snapshots": self.lakehouse.max_snapshots,
+ "max_partitions": self.lakehouse.max_partitions,
+ "max_variant_sample_rows": (
+ self.lakehouse.max_variant_sample_rows
+ ),
+ "max_variant_paths": self.lakehouse.max_variant_paths,
+ },
"database": {
"host": self.database.host,
"hosts": self.database.hosts,
@@ -2034,6 +2095,30 @@ class DorisConfig:
"Governance recent lineage event window must be in the range "
"1-525600 minutes"
)
+ if not 1 <= self.lakehouse.max_catalog_objects <= 500:
+ errors.append(
+ "Lakehouse catalog object limit must be in the range 1-500"
+ )
+ if not 1 <= self.lakehouse.max_catalog_databases <= 100:
+ errors.append(
+ "Lakehouse catalog database limit must be in the range 1-100"
+ )
+ if not 1 <= self.lakehouse.max_snapshots <= 500:
+ errors.append(
+ "Lakehouse snapshot limit must be in the range 1-500"
+ )
+ if not 1 <= self.lakehouse.max_partitions <= 1000:
+ errors.append(
+ "Lakehouse partition limit must be in the range 1-1000"
+ )
+ if not 1 <= self.lakehouse.max_variant_sample_rows <= 500:
+ errors.append(
+ "Lakehouse Variant sample limit must be in the range 1-500"
+ )
+ if not 1 <= self.lakehouse.max_variant_paths <= 2000:
+ errors.append(
+ "Lakehouse Variant path limit must be in the range 1-2000"
+ )
raw_tool_providers: Any = self.mcp_tool_providers
if not isinstance(raw_tool_providers, list):
diff --git a/doris_mcp_server/utils/lakehouse_runtime.py
b/doris_mcp_server/utils/lakehouse_runtime.py
new file mode 100644
index 0000000..35e161f
--- /dev/null
+++ b/doris_mcp_server/utils/lakehouse_runtime.py
@@ -0,0 +1,1722 @@
+# 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.
+
+"""Bounded, read-only runtime for external lakehouse and Variant evidence."""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from collections import Counter, defaultdict
+from collections.abc import Mapping, Sequence
+from datetime import date, datetime
+from typing import Any
+
+from ..tools.doris_version import (
+ DORIS_VERSION_COMMENT_QUERY,
+ DorisVersion,
+ parse_doris_version_comment,
+)
+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
+_DEFAULT_CATALOG_OBJECTS = 50
+_DEFAULT_CATALOG_DATABASES = 20
+_DEFAULT_SNAPSHOTS = 50
+_DEFAULT_PARTITIONS = 100
+_DEFAULT_VARIANT_SAMPLE_ROWS = 20
+_DEFAULT_VARIANT_PATHS = 200
+_ABSOLUTE_CATALOG_OBJECTS = 500
+_ABSOLUTE_CATALOG_DATABASES = 100
+_ABSOLUTE_SNAPSHOTS = 500
+_ABSOLUTE_PARTITIONS = 1_000
+_ABSOLUTE_VARIANT_SAMPLE_ROWS = 500
+_ABSOLUTE_VARIANT_PATHS = 2_000
+_MAX_COLUMNS = 1_000
+_MAX_PROPERTY_KEYS = 128
+_MAX_TYPED_PATHS = 256
+_DORIS_4_1 = parse_doris_version_comment("Doris version doris-4.1.0")
+_LAKEHOUSE_FORMATS = frozenset({"iceberg", "hudi", "paimon", "delta"})
+_SNAPSHOT_SYSTEM_TABLE_FORMATS = frozenset({"iceberg", "paimon"})
+_SYSTEM_TABLE_SUFFIXES = frozenset({"snapshots", "partitions"})
+_SENSITIVE_PROPERTY_PARTS = (
+ "password",
+ "secret",
+ "token",
+ "credential",
+ "keytab",
+ "access_key",
+ "secret_key",
+ "private_key",
+ "jdbc_url",
+ "endpoint",
+ "uri",
+)
+_SAFE_VARIANT_PROPERTIES = frozenset(
+ {
+ "storage_format",
+ "variant_doc_hash_shard_count",
+ "variant_doc_materialization_min_rows",
+ "variant_enable_doc_mode",
+ "variant_enable_typed_paths_to_sparse",
+ "variant_flatten_nested",
+ "variant_max_sparse_column_statistics_size",
+ "variant_max_subcolumns_count",
+ "variant_sparse_hash_shard_count",
+ }
+)
+_CATALOG_CAPABILITIES: Mapping[str, tuple[str, ...]] = {
+ "iceberg": (
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ "snapshots",
+ "system_tables",
+ "time_travel",
+ ),
+ "paimon": (
+ "incremental_read",
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ "snapshots",
+ "system_tables",
+ "time_travel",
+ ),
+ "hudi": (
+ "incremental_read",
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ "time_travel",
+ ),
+ "hms": (
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ ),
+ "hive": (
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ ),
+ "delta": (
+ "metadata_discovery",
+ "partition_pruning",
+ "predicate_pushdown",
+ "time_travel",
+ ),
+ "jdbc": ("metadata_discovery", "predicate_pushdown"),
+ "es": ("metadata_discovery", "predicate_pushdown"),
+}
+_SIMPLE_PATH_SEGMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_-]{0,254}")
+_VARIANT_PROPERTY = re.compile(
+ r"""(?P<key>storage_format|variant_[a-z0-9_]+)
+ ["']?\s*=\s*["'](?P<value>[^"']{0,256})["']""",
+ re.IGNORECASE | re.VERBOSE,
+)
+_TYPED_PATH = re.compile(
+ r"""["'](?P<path>[^"']{1,255})["']\s*:\s*
+ (?P<type>[A-Za-z][A-Za-z0-9_]*(?:\s*\([^)]*\))?)""",
+ re.VERBOSE,
+)
+
+
+class LakehouseRuntimeFailure(RuntimeError):
+ """Sanitized failure carrying a stable Lakehouse 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 DorisLakehouseRuntime:
+ """Inspect external lakehouse metadata without exposing source secrets."""
+
+ def __init__(self, connection_manager: DorisConnectionManager) -> None:
+ self._connection_manager = connection_manager
+ config = getattr(
+ getattr(connection_manager, "config", None),
+ "lakehouse",
+ None,
+ )
+ self._max_catalog_objects = _configured_limit(
+ getattr(config, "max_catalog_objects", None),
+ default=_DEFAULT_CATALOG_OBJECTS,
+ absolute=_ABSOLUTE_CATALOG_OBJECTS,
+ )
+ self._max_catalog_databases = _configured_limit(
+ getattr(config, "max_catalog_databases", None),
+ default=_DEFAULT_CATALOG_DATABASES,
+ absolute=_ABSOLUTE_CATALOG_DATABASES,
+ )
+ self._max_snapshots = _configured_limit(
+ getattr(config, "max_snapshots", None),
+ default=_DEFAULT_SNAPSHOTS,
+ absolute=_ABSOLUTE_SNAPSHOTS,
+ )
+ self._max_partitions = _configured_limit(
+ getattr(config, "max_partitions", None),
+ default=_DEFAULT_PARTITIONS,
+ absolute=_ABSOLUTE_PARTITIONS,
+ )
+ self._max_variant_sample_rows = _configured_limit(
+ getattr(config, "max_variant_sample_rows", None),
+ default=_DEFAULT_VARIANT_SAMPLE_ROWS,
+ absolute=_ABSOLUTE_VARIANT_SAMPLE_ROWS,
+ )
+ self._max_variant_paths = _configured_limit(
+ getattr(config, "max_variant_paths", None),
+ default=_DEFAULT_VARIANT_PATHS,
+ absolute=_ABSOLUTE_VARIANT_PATHS,
+ )
+ self._session_prefix = f"lakehouse_{uuid.uuid4().hex[:8]}"
+
+ async def inspect_external_catalog(
+ self,
+ *,
+ catalog: str,
+ include_objects: bool = False,
+ object_limit: int | None = None,
+ ) -> dict[str, Any]:
+ """Return sanitized catalog metadata and an optional bounded sample."""
+ catalog_name = _identifier(catalog, "catalog name")
+ limit = _bounded_int(
+ object_limit,
+ default=min(_DEFAULT_CATALOG_OBJECTS, self._max_catalog_objects),
+ maximum=self._max_catalog_objects,
+ field="object limit",
+ )
+ (
+ catalog_item,
+ properties,
+ evidence,
+ catalog_warning,
+ ) = await self._external_catalog(catalog_name)
+ catalog_type = _canonical_catalog_type(catalog_item.get("type"))
+ warnings = [catalog_warning] if catalog_warning else []
+ object_sample: dict[str, Any] = {
+ "databases": [],
+ "relations": [],
+ "truncated": False,
+ }
+ if include_objects:
+ object_sample, object_evidence, object_warnings = (
+ await self._catalog_object_sample(
+ catalog_name,
+ relation_limit=limit,
+ )
+ )
+ evidence.extend(object_evidence)
+ warnings.extend(object_warnings)
+
+ property_summary = _catalog_property_summary(properties)
+ data = {
+ "catalog": catalog_name,
+ "scope": "external",
+ "reported_type": catalog_type,
+ "visibility": "visible_to_caller",
+ "is_current": _as_bool(catalog_item.get("is_current")),
+ "created_at": _json_value(catalog_item.get("create_time")),
+ "last_updated_at": _json_value(
+ catalog_item.get("last_update_time")
+ ),
+ "declared_capabilities": list(
+ _CATALOG_CAPABILITIES.get(
+ catalog_type,
+ ("metadata_discovery",),
+ )
+ ),
+ "capability_evidence_quality": "inferred_from_reported_type",
+ "configuration": property_summary,
+ "object_sample": object_sample,
+ }
+ return _result(
+ data,
+ source="SHOW CATALOGS + SHOW CATALOG",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={
+ "property_values_returned": False,
+ "object_limit": limit,
+ },
+ )
+
+ async def inspect_lakehouse_table(
+ self,
+ *,
+ catalog: str,
+ database: str,
+ table: str,
+ include_snapshots: bool = False,
+ include_partitions: bool = False,
+ ) -> dict[str, Any]:
+ """Inspect one external table with format-specific recorded
evidence."""
+ catalog_name = _identifier(catalog, "catalog name")
+ database_name = _identifier(database, "database name")
+ table_name = _identifier(table, "table name")
+ (
+ catalog_item,
+ properties,
+ evidence,
+ catalog_warning,
+ ) = await self._external_catalog(catalog_name)
+ del properties
+ warnings = [catalog_warning] if catalog_warning else []
+ relation = build_table_reference(
+ table_name,
+ database_name,
+ catalog_name,
+ )
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _execute sends this read-only metadata command.
+ columns = await self._execute(
+ f"SHOW FULL COLUMNS FROM {relation}", # nosec B608
+ max_rows=_MAX_COLUMNS,
+ )
+ if not columns:
+ raise LakehouseRuntimeFailure(
+ "The requested external table is unavailable.",
+ reason_code="LAKEHOUSE_TABLE_NOT_FOUND",
+ status_code=404,
+ )
+ evidence.append(
+ _evidence("SHOW FULL COLUMNS", success=True, rows=len(columns))
+ )
+
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _optional_execute reaches the read-only SQL sink.
+ create_rows, create_failure = await self._optional_execute(
+ f"SHOW CREATE TABLE {relation}", # nosec B608
+ max_rows=1,
+ )
+ create_sql = _create_statement(create_rows)
+ evidence.append(
+ _evidence(
+ "SHOW CREATE TABLE",
+ success=create_failure is None,
+ rows=len(create_rows),
+ failure=create_failure,
+ )
+ )
+ if create_failure is not None:
+ warnings.append(
+ "External table DDL metadata is unavailable; format evidence "
+ "is limited to the catalog provider."
+ )
+
+ catalog_type = _canonical_catalog_type(catalog_item.get("type"))
+ table_format = _infer_table_format(catalog_type, create_sql)
+ if table_format not in _LAKEHOUSE_FORMATS:
+ raise LakehouseRuntimeFailure(
+ "The requested table is not exposed as a supported lakehouse
format.",
+ reason_code="LAKEHOUSE_TABLE_FORMAT_UNSUPPORTED",
+ status_code=409,
+ )
+
+ stats, stats_failure = await self._table_statistics(relation)
+ evidence.append(
+ _evidence(
+ "SHOW TABLE STATS",
+ success=stats_failure is None,
+ rows=1 if stats else 0,
+ failure=stats_failure,
+ )
+ )
+ if stats_failure is not None:
+ warnings.append("External table statistics are unavailable.")
+
+ snapshots: list[dict[str, Any]] = []
+ snapshots_truncated = False
+ if include_snapshots:
+ (
+ snapshots,
+ snapshots_truncated,
+ snapshot_evidence,
+ snapshot_warning,
+ ) = await self._snapshot_metadata(
+ catalog_name,
+ database_name,
+ table_name,
+ table_format,
+ )
+ evidence.append(snapshot_evidence)
+ if snapshot_warning:
+ warnings.append(snapshot_warning)
+
+ partitions: list[dict[str, Any]] = []
+ partitions_truncated = False
+ if include_partitions:
+ (
+ partitions,
+ partitions_truncated,
+ partition_evidence,
+ partition_warning,
+ ) = await self._partition_metadata(
+ catalog_name,
+ database_name,
+ table_name,
+ table_format,
+ )
+ evidence.append(partition_evidence)
+ if partition_warning:
+ warnings.append(partition_warning)
+
+ plan_facets, plan_evidence, plan_warning = await self._plan_facets(
+ relation
+ )
+ evidence.append(plan_evidence)
+ if plan_warning:
+ warnings.append(plan_warning)
+ version = await self._version()
+ lifecycle = _lifecycle_capabilities(
+ version,
+ table_format,
+ columns,
+ snapshots,
+ )
+ if not plan_facets["predicate_pushdown_observed"]:
+ warnings.append(
+ "The metadata-only probe has no caller predicate and therefore
"
+ "does not prove filter pushdown for a business query."
+ )
+
+ data = {
+ "catalog": catalog_name,
+ "database": database_name,
+ "table": table_name,
+ "format": table_format,
+ "format_evidence_quality": (
+ "reported"
+ if catalog_type in _LAKEHOUSE_FORMATS
+ else "inferred_from_sanitized_ddl"
+ ),
+ "columns": [_column_item(row) for row in columns],
+ "column_count": len(columns),
+ "partition_columns": _partition_columns(create_sql),
+ "statistics": stats,
+ "snapshots": {
+ "items": snapshots,
+ "truncated": snapshots_truncated,
+ "supported_for_format": (
+ table_format in _SNAPSHOT_SYSTEM_TABLE_FORMATS
+ ),
+ },
+ "partitions": {
+ "items": partitions,
+ "truncated": partitions_truncated,
+ },
+ "pushdown": plan_facets,
+ "lifecycle": lifecycle,
+ }
+ return _result(
+ data,
+ source="external catalog metadata",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={
+ "raw_ddl_returned": False,
+ "raw_plan_returned": False,
+ "storage_locations_returned": False,
+ },
+ )
+
+ async def inspect_variant_column(
+ self,
+ *,
+ catalog: str | None,
+ database: str,
+ table: str,
+ column: str,
+ path: str | None = None,
+ sample_rows: int | None = None,
+ ) -> dict[str, Any]:
+ """Inspect Variant configuration and sampled type shape without
values."""
+ catalog_name = (
+ _identifier(catalog, "catalog name")
+ if catalog is not None
+ else "internal"
+ )
+ database_name = _identifier(database, "database name")
+ table_name = _identifier(table, "table name")
+ column_name = _identifier(column, "column name")
+ sample_limit = _bounded_int(
+ sample_rows,
+ default=min(
+ _DEFAULT_VARIANT_SAMPLE_ROWS,
+ self._max_variant_sample_rows,
+ ),
+ maximum=self._max_variant_sample_rows,
+ field="sample rows",
+ )
+ path_segments = _variant_path(path)
+ relation = build_table_reference(
+ table_name,
+ database_name,
+ None if catalog_name == "internal" else catalog_name,
+ )
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _execute sends this read-only metadata command.
+ columns = await self._execute(
+ f"SHOW FULL COLUMNS FROM {relation}", # nosec B608
+ max_rows=_MAX_COLUMNS,
+ )
+ target = next(
+ (
+ row
+ for row in columns
+ if str(_value(row, "field", "column_name") or "") ==
column_name
+ ),
+ None,
+ )
+ if target is None:
+ raise LakehouseRuntimeFailure(
+ "The requested Variant column is unavailable.",
+ reason_code="LAKEHOUSE_VARIANT_COLUMN_NOT_FOUND",
+ status_code=404,
+ )
+ declared_type = str(_value(target, "type", "data_type") or "")
+ if not declared_type.upper().startswith("VARIANT"):
+ raise LakehouseRuntimeFailure(
+ "The requested column is not a Doris Variant column.",
+ reason_code="LAKEHOUSE_COLUMN_NOT_VARIANT",
+ status_code=409,
+ )
+
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _optional_execute reaches the read-only SQL sink.
+ create_rows, create_failure = await self._optional_execute(
+ f"SHOW CREATE TABLE {relation}", # nosec B608
+ max_rows=1,
+ )
+ create_sql = _create_statement(create_rows)
+ properties = _variant_properties(f"{declared_type}\n{create_sql}")
+ configuration, configuration_warnings = _variant_configuration(
+ properties
+ )
+
+ column_ref = quote_identifier(column_name, "column name")
+ expression = column_ref
+ params: list[Any] = []
+ for segment in path_segments:
+ expression += "[%s]"
+ params.append(segment)
+ # SQL sink audit: relation and column are validated/quoted, every path
+ # segment remains driver-bound, and _execute receives a bounded limit.
+ sample_sql = (
+ f"SELECT VARIANT_TYPE({expression}) AS `variant_type` " # nosec
B608
+ f"FROM {relation} WHERE {column_ref} IS NOT NULL "
+ f"LIMIT {sample_limit}"
+ )
+ sample_result, sample_failure = await self._optional_execute(
+ sample_sql,
+ params=tuple(params),
+ max_rows=sample_limit,
+ )
+ warnings = list(configuration_warnings)
+ if create_failure is not None:
+ warnings.append(
+ "Table DDL is unavailable; Variant mode properties may be
incomplete."
+ )
+ if sample_failure is not None:
+ warnings.append(
+ "Bounded VARIANT_TYPE sampling is unavailable; no row values "
+ "or inferred path shapes were returned."
+ )
+ observed_paths, paths_truncated = _variant_type_observations(
+ sample_result,
+ requested_path=path,
+ maximum=self._max_variant_paths,
+ )
+ version = await self._version()
+ advanced = _variant_advanced_capabilities(version, configuration)
+ typed_paths = _typed_paths(declared_type)
+ if len(typed_paths) >= _MAX_TYPED_PATHS:
+ warnings.append(
+ f"Typed Variant path metadata was limited to
{_MAX_TYPED_PATHS} paths."
+ )
+ if paths_truncated:
+ warnings.append(
+ f"Observed Variant paths were limited to
{self._max_variant_paths}."
+ )
+
+ evidence = [
+ _evidence("SHOW FULL COLUMNS", success=True, rows=len(columns)),
+ _evidence(
+ "SHOW CREATE TABLE",
+ success=create_failure is None,
+ rows=len(create_rows),
+ failure=create_failure,
+ ),
+ _evidence(
+ "VARIANT_TYPE bounded sample",
+ success=sample_failure is None,
+ rows=len(sample_result),
+ failure=sample_failure,
+ ),
+ ]
+ data = {
+ "catalog": catalog_name,
+ "database": database_name,
+ "table": table_name,
+ "column": column_name,
+ "declared_type": declared_type,
+ "nullable": str(_value(target, "null") or "").upper() == "YES",
+ "requested_path": path,
+ "typed_paths": typed_paths,
+ "configuration": configuration,
+ "advanced_capabilities": advanced,
+ "shape_sample": {
+ "rows_observed": len(sample_result),
+ "paths": observed_paths,
+ "truncated": paths_truncated,
+ },
+ }
+ return _result(
+ data,
+ source="SHOW FULL COLUMNS + VARIANT_TYPE",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={
+ "sample_limit": sample_limit,
+ "sampled_values_returned": False,
+ "raw_ddl_returned": False,
+ },
+ )
+
+ async def _external_catalog(
+ self,
+ catalog_name: str,
+ ) -> tuple[
+ dict[str, Any],
+ dict[str, Any],
+ list[dict[str, Any]],
+ str | None,
+ ]:
+ rows = await self._execute("SHOW CATALOGS", max_rows=1_000)
+ catalog_item = next(
+ (
+ _catalog_item(row)
+ for row in rows
+ if str(_value(row, "catalogname", "catalog_name") or "")
+ == catalog_name
+ ),
+ None,
+ )
+ if catalog_item is None:
+ raise LakehouseRuntimeFailure(
+ "The requested external catalog is unavailable.",
+ reason_code="LAKEHOUSE_CATALOG_NOT_FOUND",
+ status_code=404,
+ )
+ if catalog_name == "internal":
+ raise LakehouseRuntimeFailure(
+ "The internal Doris catalog is not an external catalog.",
+ reason_code="LAKEHOUSE_CATALOG_NOT_EXTERNAL",
+ status_code=409,
+ )
+ catalog_ref = quote_identifier(catalog_name, "catalog name")
+ # SQL sink audit: catalog_ref is a strictly validated and quoted
+ # identifier before _optional_execute sends this read-only command.
+ property_rows, property_failure = await self._optional_execute(
+ f"SHOW CATALOG {catalog_ref}", # nosec B608
+ max_rows=1_000,
+ )
+ properties = {
+ str(key).strip().casefold(): value
+ for row in property_rows
+ if (key := _value(row, "key", "property", "name")) is not None
+ for value in (_value(row, "value", "property_value"),)
+ }
+ return (
+ catalog_item,
+ properties,
+ [
+ _evidence("SHOW CATALOGS", success=True, rows=len(rows)),
+ _evidence(
+ "SHOW CATALOG",
+ success=property_failure is None,
+ rows=len(property_rows),
+ failure=property_failure,
+ ),
+ ],
+ (
+ None
+ if property_failure is None
+ else "External catalog configuration metadata is unavailable."
+ ),
+ )
+
+ async def _catalog_object_sample(
+ self,
+ catalog_name: str,
+ *,
+ relation_limit: int,
+ ) -> tuple[dict[str, Any], list[dict[str, Any]], list[str]]:
+ catalog_ref = quote_identifier(catalog_name, "catalog name")
+ # SQL sink audit: catalog_ref is a strictly validated and quoted
+ # identifier before _optional_execute sends this read-only command.
+ database_rows, database_failure = await self._optional_execute(
+ f"SHOW DATABASES FROM {catalog_ref}", # nosec B608
+ max_rows=self._max_catalog_databases + 1,
+ )
+ if database_failure is not None:
+ return (
+ {
+ "databases": [],
+ "relations": [],
+ "truncated": False,
+ },
+ [
+ _evidence(
+ "SHOW DATABASES",
+ success=False,
+ rows=0,
+ failure=database_failure,
+ )
+ ],
+ ["Catalog database metadata is unavailable."],
+ )
+ database_names: list[str] = []
+ unaddressable_databases = 0
+ for row in database_rows:
+ value = _first_value(row)
+ if value is None:
+ continue
+ try:
+ database_names.append(
+ _identifier(str(value), "database name")
+ )
+ except LakehouseRuntimeFailure:
+ unaddressable_databases += 1
+ databases_truncated = len(database_names) > self._max_catalog_databases
+ database_names = database_names[: self._max_catalog_databases]
+ relations: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ if unaddressable_databases:
+ warnings.append(
+ f"{unaddressable_databases} catalog database names were not "
+ "addressable under the strict identifier policy."
+ )
+ evidence = [
+ _evidence(
+ "SHOW DATABASES",
+ success=True,
+ rows=len(database_rows),
+ )
+ ]
+ for database_name in database_names:
+ if len(relations) >= relation_limit:
+ break
+ scope = (
+ f"{quote_identifier(catalog_name, 'catalog name')}."
+ f"{quote_identifier(database_name, 'database name')}"
+ )
+ # SQL sink audit: scope contains only strictly validated and quoted
+ # identifiers before _optional_execute reaches the read-only sink.
+ rows, failure = await self._optional_execute(
+ f"SHOW FULL TABLES FROM {scope}", # nosec B608
+ max_rows=(relation_limit - len(relations)) + 1,
+ )
+ evidence.append(
+ _evidence(
+ "SHOW FULL TABLES",
+ success=failure is None,
+ rows=len(rows),
+ failure=failure,
+ )
+ )
+ if failure is not None:
+ warnings.append(
+ f"Relation metadata is unavailable for database
{database_name}."
+ )
+ continue
+ for row in rows:
+ if len(relations) >= relation_limit:
+ break
+ name = _table_name(row)
+ if name is None:
+ continue
+ relations.append(
+ {
+ "database": database_name,
+ "name": name,
+ "type": _table_type(row),
+ }
+ )
+ truncated = (
+ databases_truncated
+ or bool(unaddressable_databases)
+ or len(relations) >= relation_limit
+ )
+ return (
+ {
+ "databases": [{"name": name} for name in database_names],
+ "relations": relations,
+ "truncated": truncated,
+ },
+ evidence,
+ warnings,
+ )
+
+ async def _snapshot_metadata(
+ self,
+ catalog: str,
+ database: str,
+ table: str,
+ table_format: str,
+ ) -> tuple[
+ list[dict[str, Any]],
+ bool,
+ dict[str, Any],
+ str | None,
+ ]:
+ if table_format not in _SNAPSHOT_SYSTEM_TABLE_FORMATS:
+ return (
+ [],
+ False,
+ _evidence(
+ "format system table",
+ success=False,
+ rows=0,
+ reason_code="SNAPSHOT_SYSTEM_TABLE_NOT_SUPPORTED",
+ ),
+ f"Snapshot system-table inspection is not defined for
{table_format}.",
+ )
+ reference = _system_table_reference(
+ catalog,
+ database,
+ table,
+ "snapshots",
+ )
+ # SQL sink audit: reference is assembled from validated/quoted user
+ # identifiers plus a fixed suffix before _optional_execute reaches
Doris.
+ rows, failure = await self._optional_execute(
+ f"SELECT * FROM {reference} LIMIT {self._max_snapshots + 1}", #
nosec B608
+ max_rows=self._max_snapshots + 1,
+ )
+ if failure is not None:
+ return (
+ [],
+ False,
+ _evidence(
+ f"{table_format} snapshots system table",
+ success=False,
+ rows=0,
+ failure=failure,
+ ),
+ "Snapshot metadata is unavailable for the requested table.",
+ )
+ truncated = len(rows) > self._max_snapshots
+ return (
+ [_snapshot_item(row) for row in rows[: self._max_snapshots]],
+ truncated,
+ _evidence(
+ f"{table_format} snapshots system table",
+ success=True,
+ rows=len(rows),
+ ),
+ None,
+ )
+
+ async def _partition_metadata(
+ self,
+ catalog: str,
+ database: str,
+ table: str,
+ table_format: str,
+ ) -> tuple[
+ list[dict[str, Any]],
+ bool,
+ dict[str, Any],
+ str | None,
+ ]:
+ relation = build_table_reference(table, database, catalog)
+ # SQL sink audit: relation contains only validated and quoted
identifiers,
+ # and the integer limit is bounded before _optional_execute reaches
Doris.
+ rows, failure = await self._optional_execute(
+ f"SHOW PARTITIONS FROM {relation} LIMIT {self._max_partitions +
1}", # nosec B608
+ max_rows=self._max_partitions + 1,
+ )
+ source = "SHOW PARTITIONS"
+ if failure is not None and table_format in
_SNAPSHOT_SYSTEM_TABLE_FORMATS:
+ reference = _system_table_reference(
+ catalog,
+ database,
+ table,
+ "partitions",
+ )
+ # SQL sink audit: reference is assembled from validated/quoted user
+ # identifiers plus a fixed suffix and a bounded integer limit.
+ rows, failure = await self._optional_execute(
+ f"SELECT * FROM {reference} LIMIT {self._max_partitions + 1}",
# nosec B608
+ max_rows=self._max_partitions + 1,
+ )
+ source = f"{table_format} partitions system table"
+ if failure is not None:
+ return (
+ [],
+ False,
+ _evidence(
+ source,
+ success=False,
+ rows=0,
+ failure=failure,
+ ),
+ "Partition metadata is unavailable for the requested table.",
+ )
+ truncated = len(rows) > self._max_partitions
+ return (
+ [_partition_item(row) for row in rows[: self._max_partitions]],
+ truncated,
+ _evidence(source, success=True, rows=len(rows)),
+ None,
+ )
+
+ async def _table_statistics(
+ self,
+ relation: str,
+ ) -> tuple[dict[str, Any], LakehouseRuntimeFailure | None]:
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _optional_execute reaches the read-only SQL sink.
+ rows, failure = await self._optional_execute(
+ f"SHOW TABLE STATS {relation}", # nosec B608
+ max_rows=2,
+ )
+ if not rows:
+ return {}, failure
+ row = rows[0]
+ return (
+ {
+ "row_count": _number(
+ _value(row, "row_count", "rowcount", "rows")
+ ),
+ "data_size_bytes": _number(
+ _value(
+ row,
+ "data_size",
+ "data_length",
+ "datasize",
+ )
+ ),
+ "updated_at": _json_value(
+ _value(row, "update_time", "updated_at")
+ ),
+ },
+ failure,
+ )
+
+ async def _plan_facets(
+ self,
+ relation: str,
+ ) -> tuple[dict[str, Any], dict[str, Any], str | None]:
+ # SQL sink audit: relation contains only strictly validated and quoted
+ # identifiers before _optional_execute reaches this read-only EXPLAIN.
+ rows, failure = await self._optional_execute(
+ f"EXPLAIN SELECT * FROM {relation} LIMIT 1", # nosec B608
+ max_rows=512,
+ )
+ if failure is not None:
+ return (
+ {
+ "external_scan_observed": False,
+ "partition_pruning_observed": False,
+ "predicate_pushdown_observed": False,
+ "scan_nodes": [],
+ },
+ _evidence(
+ "EXPLAIN",
+ success=False,
+ rows=0,
+ failure=failure,
+ ),
+ "External scan plan evidence is unavailable.",
+ )
+ plan = "\n".join(
+ str(value)
+ for row in rows
+ for value in row.values()
+ if value is not None
+ )
+ upper = plan.upper()
+ scan_nodes = sorted(
+ {
+ marker
+ for marker in (
+ "HIVE_SCAN_NODE",
+ "ICEBERG_SCAN_NODE",
+ "HUDI_SCAN_NODE",
+ "PAIMON_SCAN_NODE",
+ "JDBC_SCAN_NODE",
+ "ES_SCAN_NODE",
+ "FILE_SCAN_NODE",
+ "VOlapScanNode",
+ )
+ if marker.upper() in upper
+ }
+ )
+ return (
+ {
+ "external_scan_observed": bool(scan_nodes)
+ or "EXTERNAL" in upper,
+ "partition_pruning_observed": any(
+ marker in upper
+ for marker in (
+ "PARTITION PREDICATES",
+ "PARTITIONS=",
+ "PARTITION PRUNING",
+ )
+ ),
+ "predicate_pushdown_observed": any(
+ marker in upper
+ for marker in (
+ "PUSHDOWN",
+ "PREDICATES:",
+ "PUSH DOWN",
+ )
+ ),
+ "scan_nodes": scan_nodes,
+ },
+ _evidence("EXPLAIN", success=True, rows=len(rows)),
+ None,
+ )
+
+ async def _version(self) -> DorisVersion:
+ rows, failure = await self._optional_execute(
+ DORIS_VERSION_COMMENT_QUERY,
+ max_rows=1,
+ )
+ if failure is not None or not rows:
+ return parse_doris_version_comment("")
+ value = _value(
+ rows[0],
+ "version_comment",
+ "@@version_comment",
+ )
+ return parse_doris_version_comment(
+ str(value) if value is not None else ""
+ )
+
+ async def _optional_execute(
+ self,
+ sql: str,
+ *,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ max_rows: int,
+ ) -> tuple[list[dict[str, Any]], LakehouseRuntimeFailure | None]:
+ try:
+ return (
+ await self._execute(
+ sql,
+ params=params,
+ max_rows=max_rows,
+ ),
+ None,
+ )
+ except LakehouseRuntimeFailure as exc:
+ return [], exc
+
+ async def _execute(
+ self,
+ sql: str,
+ *,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ max_rows: int,
+ ) -> 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:
+ 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],
+ evidence: Sequence[Mapping[str, Any]],
+ metadata: Mapping[str, Any],
+) -> dict[str, Any]:
+ unique_warnings = list(dict.fromkeys(str(item) for item in warnings))
+ return {
+ "status": "partial" if unique_warnings else "success",
+ "data": redact_sensitive_data(dict(data)),
+ "warnings": unique_warnings,
+ "metadata": {
+ "source": source,
+ **dict(metadata),
+ },
+ "evidence": [
+ redact_sensitive_data(dict(item))
+ for item in evidence
+ ],
+ }
+
+
+def _evidence(
+ source: str,
+ *,
+ success: bool,
+ rows: int,
+ failure: LakehouseRuntimeFailure | None = None,
+ reason_code: str | None = None,
+) -> dict[str, Any]:
+ return {
+ "source": source,
+ "success": success,
+ "rows_observed": rows,
+ "reason_code": (
+ reason_code
+ if reason_code is not None
+ else (None if failure is None else failure.reason_code)
+ ),
+ }
+
+
+def _identifier(value: Any, field: str) -> str:
+ try:
+ return validate_identifier(value, field)
+ except (SQLSecurityError, TypeError) as exc:
+ raise _argument_failure(f"Invalid {field}.") from exc
+
+
+def _bounded_int(
+ value: Any,
+ *,
+ default: int,
+ maximum: int,
+ field: str,
+) -> int:
+ if value is None:
+ return default
+ if isinstance(value, bool) or not isinstance(value, int):
+ raise _argument_failure(f"{field.capitalize()} must be an integer.")
+ if not 1 <= value <= maximum:
+ raise _argument_failure(
+ f"{field.capitalize()} must be between 1 and {maximum}."
+ )
+ return int(value)
+
+
+def _configured_limit(value: Any, *, default: int, absolute: int) -> int:
+ if isinstance(value, bool) or not isinstance(value, int):
+ return default
+ return value if 1 <= value <= absolute else default
+
+
+def _variant_path(value: Any) -> tuple[str, ...]:
+ if value is None:
+ return ()
+ if not isinstance(value, str):
+ raise _argument_failure("Variant path must be a string.")
+ normalized = value.strip()
+ if not normalized or len(normalized) > 1_024:
+ raise _argument_failure(
+ "Variant path must contain 1-1024 characters."
+ )
+ if normalized == "$":
+ return ()
+ if normalized.startswith("$."):
+ normalized = normalized[2:]
+ elif normalized.startswith("$"):
+ raise _argument_failure(
+ "Variant path must use simple $.segment notation."
+ )
+ parts = normalized.split(".")
+ if not 1 <= len(parts) <= 16 or any(
+ _SIMPLE_PATH_SEGMENT.fullmatch(part) is None for part in parts
+ ):
+ raise _argument_failure(
+ "Variant path must contain 1-16 simple dot-separated segments."
+ )
+ return tuple(parts)
+
+
+def _system_table_reference(
+ catalog: str,
+ database: str,
+ table: str,
+ suffix: str,
+) -> str:
+ catalog_name = _identifier(catalog, "catalog name")
+ database_name = _identifier(database, "database name")
+ table_name = _identifier(table, "table name")
+ if suffix not in _SYSTEM_TABLE_SUFFIXES:
+ raise _argument_failure("Unsupported lakehouse system-table suffix.")
+ system_table = f"`{table_name}${suffix}`"
+ return (
+ f"{quote_identifier(catalog_name, 'catalog name')}."
+ f"{quote_identifier(database_name, 'database name')}."
+ f"{system_table}"
+ )
+
+
+def _catalog_item(row: Mapping[str, Any]) -> dict[str, Any]:
+ return {
+ "name": _value(row, "catalogname", "catalog_name"),
+ "type": _value(row, "type", "catalog_type"),
+ "is_current": _value(row, "iscurrent", "is_current"),
+ "create_time": _value(row, "createtime", "create_time"),
+ "last_update_time": _value(
+ row,
+ "lastupdatetime",
+ "last_update_time",
+ ),
+ "comment": _value(row, "comment"),
+ }
+
+
+def _catalog_property_summary(
+ properties: Mapping[str, Any],
+) -> dict[str, Any]:
+ safe_keys = sorted(
+ key
+ for key in properties
+ if not _sensitive_property(key)
+ )[:_MAX_PROPERTY_KEYS]
+ redacted_count = sum(
+ 1 for key in properties if _sensitive_property(key)
+ )
+ cache_keys = (
+ "use_meta_cache",
+ "metadata_cache_enabled",
+ "enable_meta_cache",
+ )
+ refresh_keys = (
+ "metadata_refresh_interval_sec",
+ "refresh_interval_sec",
+ )
+ return {
+ "property_count": len(properties),
+ "property_keys": safe_keys,
+ "property_keys_truncated": len(safe_keys)
+ < (len(properties) - redacted_count),
+ "sensitive_property_count": redacted_count,
+ "property_values_returned": False,
+ "metadata_cache_configured": any(
+ key in properties for key in cache_keys
+ ),
+ "refresh_interval_configured": any(
+ key in properties for key in refresh_keys
+ ),
+ "warehouse_configured": any(
+ key in properties
+ for key in (
+ "warehouse",
+ "warehouse_location",
+ "iceberg.catalog.warehouse",
+ )
+ ),
+ }
+
+
+def _sensitive_property(key: str) -> bool:
+ normalized = key.casefold().replace(".", "_").replace("-", "_")
+ return any(part in normalized for part in _SENSITIVE_PROPERTY_PARTS)
+
+
+def _canonical_catalog_type(value: Any) -> str:
+ normalized = (
+ str(value or "unknown")
+ .strip()
+ .casefold()
+ .replace("-", "_")
+ .replace(" ", "_")
+ )
+ aliases = {
+ "iceberg_catalog": "iceberg",
+ "paimon_catalog": "paimon",
+ "hudi_catalog": "hudi",
+ "hive": "hms",
+ "deltalake": "delta",
+ "delta_lake": "delta",
+ "elasticsearch": "es",
+ }
+ return aliases.get(normalized, normalized)
+
+
+def _infer_table_format(catalog_type: str, create_sql: str) -> str:
+ if catalog_type in _LAKEHOUSE_FORMATS:
+ return catalog_type
+ upper = create_sql.upper()
+ for marker, name in (
+ ("ICEBERG", "iceberg"),
+ ("PAIMON", "paimon"),
+ ("HUDI", "hudi"),
+ ("DELTA", "delta"),
+ ):
+ if marker in upper:
+ return name
+ return "unknown"
+
+
+def _column_item(row: Mapping[str, Any]) -> dict[str, Any]:
+ return {
+ "name": _json_value(_value(row, "field", "column_name")),
+ "type": _json_value(_value(row, "type", "data_type")),
+ "nullable": str(_value(row, "null") or "").upper() == "YES",
+ "key": _json_value(_value(row, "key")),
+ "comment": _bounded_output(_value(row, "comment"), 512),
+ }
+
+
+def _create_statement(rows: Sequence[Mapping[str, Any]]) -> str:
+ if not rows:
+ return ""
+ row = rows[0]
+ value = _value(row, "create table", "create_table", "create view")
+ if value is None:
+ values = list(row.values())
+ value = values[-1] if values else ""
+ return str(value or "")
+
+
+def _partition_columns(create_sql: str) -> list[str]:
+ match = re.search(
+ r"\bPARTITIONED?\s+BY\s*\((?P<columns>[^)]*)\)",
+ create_sql,
+ re.IGNORECASE | re.DOTALL,
+ )
+ if match is None:
+ return []
+ columns: list[str] = []
+ for item in match.group("columns").split(","):
+ raw = item.strip().strip("`").split()[0] if item.strip() else ""
+ try:
+ name = validate_identifier(raw, "partition column")
+ except SQLSecurityError:
+ continue
+ columns.append(name)
+ return columns[:128]
+
+
+def _snapshot_item(row: Mapping[str, Any]) -> dict[str, Any]:
+ return {
+ "snapshot_id": _json_value(
+ _value(row, "snapshot_id", "snapshotid", "id")
+ ),
+ "parent_id": _json_value(
+ _value(row, "parent_id", "parent_snapshot_id")
+ ),
+ "schema_id": _json_value(_value(row, "schema_id")),
+ "committed_at": _json_value(
+ _value(
+ row,
+ "commit_time",
+ "committed_at",
+ "timestamp",
+ "timestamp_ms",
+ )
+ ),
+ "operation": _bounded_output(
+ _value(row, "operation", "commit_kind", "kind"),
+ 64,
+ ),
+ "record_count": _number(
+ _value(row, "record_count", "total_record_count")
+ ),
+ "summary_available": _value(row, "summary") is not None,
+ }
+
+
+def _partition_item(row: Mapping[str, Any]) -> dict[str, Any]:
+ return {
+ "name": _bounded_output(
+ _value(row, "partitionname", "partition_name", "partition"),
+ 512,
+ ),
+ "partition_id": _json_value(
+ _value(row, "partitionid", "partition_id", "spec_id")
+ ),
+ "record_count": _number(
+ _value(row, "record_count", "row_count", "table_rows", "rows")
+ ),
+ "file_count": _number(
+ _value(row, "file_count", "data_file_count")
+ ),
+ "updated_at": _json_value(
+ _value(
+ row,
+ "lastconsistencychecktime",
+ "last_update_time",
+ "update_time",
+ )
+ ),
+ }
+
+
+def _lifecycle_capabilities(
+ version: DorisVersion,
+ table_format: str,
+ columns: Sequence[Mapping[str, Any]],
+ snapshots: Sequence[Mapping[str, Any]],
+) -> dict[str, Any]:
+ supports_4_1 = version.is_parsed and version.is_at_least(_DORIS_4_1)
+ column_names = {
+ str(_value(row, "field", "column_name") or "").casefold()
+ for row in columns
+ }
+ row_lineage_columns = sorted(
+ name
+ for name in (
+ "_row_id",
+ "_last_updated_sequence_number",
+ )
+ if name in column_names
+ )
+ eligible = supports_4_1 and table_format == "iceberg"
+ return {
+ "detected_server_version": version.normalized,
+ "iceberg_v3_lifecycle_eligible": eligible,
+ "deletion_vector": {
+ "server_support": eligible,
+ "target_evidence_observed": any(
+ "deletion" in str(item).casefold() for item in snapshots
+ ),
+ },
+ "row_lineage": {
+ "server_support": eligible,
+ "target_evidence_observed": bool(row_lineage_columns),
+ "observable_hidden_columns": row_lineage_columns,
+ },
+ }
+
+
+def _variant_properties(source: str) -> dict[str, str]:
+ properties: dict[str, str] = {}
+ for match in _VARIANT_PROPERTY.finditer(source):
+ key = match.group("key").casefold()
+ if key in _SAFE_VARIANT_PROPERTIES:
+ properties[key] = match.group("value").strip()
+ return properties
+
+
+def _variant_configuration(
+ properties: Mapping[str, str],
+) -> tuple[dict[str, Any], tuple[str, ...]]:
+ doc_mode = _as_bool(properties.get("variant_enable_doc_mode"))
+ typed_sparse = _as_bool(
+ properties.get("variant_enable_typed_paths_to_sparse")
+ )
+ sparse_shards = _number(
+ properties.get("variant_sparse_hash_shard_count")
+ )
+ doc_shards = _number(properties.get("variant_doc_hash_shard_count"))
+ sparse_mode = bool(typed_sparse) or (
+ isinstance(sparse_shards, int | float) and sparse_shards > 1
+ )
+ warnings: list[str] = []
+ if doc_mode and sparse_mode:
+ warnings.append(
+ "DOC mode and sparse Variant storage both appear configured; "
+ "Doris documents these modes as mutually exclusive."
+ )
+ mode = "doc" if doc_mode else "sparse" if sparse_mode else "default"
+ storage_format = properties.get("storage_format")
+ return (
+ {
+ "mode": mode,
+ "storage_format": (
+ storage_format.upper() if storage_format else None
+ ),
+ "storage_v3": bool(
+ storage_format and storage_format.casefold() == "v3"
+ ),
+ "max_subcolumns_count": _number(
+ properties.get("variant_max_subcolumns_count")
+ ),
+ "typed_paths_to_sparse": typed_sparse,
+ "sparse_hash_shard_count": sparse_shards,
+ "doc_hash_shard_count": doc_shards,
+ "doc_materialization_min_rows": _number(
+ properties.get("variant_doc_materialization_min_rows")
+ ),
+ "flatten_nested": _as_bool(
+ properties.get("variant_flatten_nested")
+ ),
+ },
+ tuple(warnings),
+ )
+
+
+def _variant_advanced_capabilities(
+ version: DorisVersion,
+ configuration: Mapping[str, Any],
+) -> dict[str, Any]:
+ supported = version.is_parsed and version.is_at_least(_DORIS_4_1)
+ return {
+ "detected_server_version": version.normalized,
+ "storage_v3_supported": supported,
+ "sparse_sharding_supported": supported,
+ "sparse_cache_supported": supported,
+ "doc_mode_supported": supported,
+ "configured_mode": configuration.get("mode"),
+ }
+
+
+def _typed_paths(declared_type: str) -> list[dict[str, str]]:
+ items: list[dict[str, str]] = []
+ for match in _TYPED_PATH.finditer(declared_type):
+ items.append(
+ {
+ "path": match.group("path"),
+ "type": match.group("type").upper(),
+ }
+ )
+ if len(items) >= _MAX_TYPED_PATHS:
+ break
+ return items
+
+
+def _variant_type_observations(
+ rows: Sequence[Mapping[str, Any]],
+ *,
+ requested_path: str | None,
+ maximum: int,
+) -> tuple[list[dict[str, Any]], bool]:
+ counts: dict[str, Counter[str]] = defaultdict(Counter)
+ observed_rows: Counter[str] = Counter()
+ for row in rows:
+ raw = _value(row, "variant_type")
+ if raw is None:
+ continue
+ parsed = _parse_variant_type(raw)
+ if isinstance(parsed, Mapping):
+ for path, value_type in parsed.items():
+ raw_path = str(path)
+ if requested_path and not raw_path:
+ path_name = requested_path
+ else:
+ path_name = raw_path or "$"
+ counts[path_name][_variant_type_label(value_type)] += 1
+ observed_rows[path_name] += 1
+ else:
+ path_name = requested_path or "$"
+ counts[path_name][_variant_type_label(parsed)] += 1
+ observed_rows[path_name] += 1
+ ordered = sorted(
+ counts,
+ key=lambda path: (-observed_rows[path], path),
+ )
+ truncated = len(ordered) > maximum
+ items = []
+ denominator = len(rows)
+ for path_name in ordered[:maximum]:
+ present = observed_rows[path_name]
+ items.append(
+ {
+ "path": path_name,
+ "types": [
+ {"type": value_type, "rows": count}
+ for value_type, count in counts[path_name].most_common()
+ ],
+ "rows_observed": present,
+ "presence_ratio": (
+ round(present / denominator, 6) if denominator else None
+ ),
+ }
+ )
+ return items, truncated
+
+
+def _variant_type_label(value: Any) -> str:
+ return str(value).strip().upper()[:128] or "UNKNOWN"
+
+
+def _parse_variant_type(value: Any) -> Mapping[str, Any] | str:
+ if isinstance(value, Mapping):
+ return value
+ text = str(value)
+ try:
+ parsed = json.loads(text)
+ except (TypeError, ValueError):
+ return text[:128]
+ if isinstance(parsed, Mapping):
+ return parsed
+ if isinstance(parsed, str):
+ return parsed[:128]
+ return type(parsed).__name__
+
+
+def _table_name(row: Mapping[str, Any]) -> str | None:
+ value = next(
+ (
+ value
+ for key, value in row.items()
+ if str(key).casefold().startswith("tables_in_")
+ ),
+ _first_value(row),
+ )
+ if value is None:
+ return None
+ try:
+ return validate_identifier(str(value), "table name")
+ except SQLSecurityError:
+ return None
+
+
+def _table_type(row: Mapping[str, Any]) -> str:
+ raw = str(_value(row, "table_type", "type") or "table").casefold()
+ return "view" if "view" in raw else "table"
+
+
+def _first_value(row: Mapping[str, Any]) -> Any:
+ return next(iter(row.values()), None)
+
+
+def _value(row: Mapping[str, Any], *names: str) -> Any:
+ normalized = {
+ str(key).strip().casefold().replace(" ", "_"): value
+ for key, value in row.items()
+ }
+ for name in names:
+ key = name.strip().casefold().replace(" ", "_")
+ if key in normalized:
+ return normalized[key]
+ return None
+
+
+def _json_value(value: Any) -> Any:
+ if isinstance(value, datetime | date):
+ return value.isoformat()
+ if isinstance(value, bytes):
+ return value.decode("utf-8", errors="replace")
+ if value is None or isinstance(value, str | int | float | bool):
+ return value
+ return str(value)
+
+
+def _number(value: Any) -> int | float | None:
+ if value in (None, "") or isinstance(value, bool):
+ return None
+ try:
+ number = float(str(value).replace(",", "").strip())
+ except (TypeError, ValueError):
+ return None
+ return int(number) if number.is_integer() else number
+
+
+def _as_bool(value: Any) -> bool:
+ if isinstance(value, bool):
+ return value
+ return str(value or "").strip().casefold() in {
+ "1",
+ "true",
+ "yes",
+ "on",
+ }
+
+
+def _bounded_output(value: Any, maximum: int) -> str | None:
+ if value is None:
+ return None
+ normalized = str(value).replace("\x00", "").replace("\r", " ").replace(
+ "\n",
+ " ",
+ )
+ return normalized[:maximum]
+
+
+def _argument_failure(message: str) -> LakehouseRuntimeFailure:
+ return LakehouseRuntimeFailure(
+ message,
+ reason_code="LAKEHOUSE_ARGUMENT_INVALID",
+ status_code=400,
+ )
+
+
+def _classify_failure(error: Exception) -> LakehouseRuntimeFailure:
+ if isinstance(error, LakehouseRuntimeFailure):
+ return error
+ if isinstance(error, SQLSecurityError):
+ return _argument_failure("Lakehouse arguments are invalid.")
+ error_code = next(
+ (item for item in getattr(error, "args", ()) if isinstance(item, int)),
+ None,
+ )
+ if error_code in {1044, 1045, 1142, 1227}:
+ return LakehouseRuntimeFailure(
+ "Doris denied access to the requested lakehouse metadata.",
+ reason_code="LAKEHOUSE_PERMISSION_DENIED",
+ status_code=403,
+ )
+ if error_code in {
+ 1049,
+ 1054,
+ 1064,
+ 1109,
+ 1146,
+ 1176,
+ 1305,
+ }:
+ return LakehouseRuntimeFailure(
+ "The requested Doris lakehouse metadata is unavailable.",
+ reason_code="LAKEHOUSE_METADATA_UNAVAILABLE",
+ status_code=404,
+ )
+ if isinstance(error, ConnectionError | TimeoutError):
+ return LakehouseRuntimeFailure(
+ "Doris lakehouse metadata is temporarily unavailable.",
+ reason_code="LAKEHOUSE_CONNECTION_FAILED",
+ status_code=503,
+ retryable=True,
+ )
+ return LakehouseRuntimeFailure(
+ "Doris lakehouse metadata execution failed.",
+ reason_code="LAKEHOUSE_EXECUTION_FAILED",
+ status_code=502,
+ )
+
+
+__all__ = [
+ "DorisLakehouseRuntime",
+ "LakehouseRuntimeFailure",
+]
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index 8d12f32..eb20e44 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -107,6 +107,11 @@ GOVERNANCE_CHILD_NAMES = (
"list_udfs",
"get_auth_mapping_status",
)
+LAKEHOUSE_CHILD_NAMES = (
+ "inspect_external_catalog",
+ "inspect_lakehouse_table",
+ "inspect_variant_column",
+)
@dataclass(frozen=True)
@@ -147,6 +152,18 @@ class DorisSearchSandbox:
return f"`{self.settings.database}`.`{self.table}`"
+@dataclass
+class DorisVariantSandbox:
+ settings: RealDorisSettings
+ admin_connection: pymysql.Connection
+ table: str
+ marker: str
+
+ @property
+ def qualified_table(self) -> str:
+ return f"`{self.settings.database}`.`{self.table}`"
+
+
def _real_doris_settings() -> RealDorisSettings:
required = {
name: os.getenv(name, "").strip()
@@ -286,6 +303,70 @@ def doris_search_sandbox() -> DorisSearchSandbox:
admin_connection.close()
[email protected]
+def doris_variant_sandbox() -> DorisVariantSandbox:
+ settings = _real_doris_settings()
+ table = f"mcp_variant_it_{secrets.token_hex(6)}"
+ marker = secrets.token_hex(12)
+ qualified_table = f"`{settings.database}`.`{table}`"
+ admin_connection = pymysql.connect(
+ host=settings.host,
+ port=settings.port,
+ user=settings.user,
+ password=settings.password,
+ database=settings.database,
+ autocommit=True,
+ )
+
+ try:
+ with admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"""
+ CREATE TABLE {qualified_table} (
+ id BIGINT,
+ payload VARIANT
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ )
+ cursor.executemany(
+ f"INSERT INTO {qualified_table} VALUES (%s, %s)",
+ [
+ (
+ 1,
+ json.dumps(
+ {
+ "profile": {"age": 42},
+ "private_marker": marker,
+ }
+ ),
+ ),
+ (
+ 2,
+ json.dumps(
+ {
+ "profile": {"age": 43},
+ "private_marker": marker,
+ }
+ ),
+ ),
+ ],
+ )
+
+ yield DorisVariantSandbox(
+ settings=settings,
+ admin_connection=admin_connection,
+ table=table,
+ marker=marker,
+ )
+ finally:
+ with suppress(Exception), admin_connection.cursor() as cursor:
+ cursor.execute(f"DROP TABLE IF EXISTS {qualified_table}")
+ admin_connection.close()
+
+
def _server_environment(
settings: RealDorisSettings,
*,
@@ -1713,6 +1794,98 @@ async def
test_real_doris_hierarchical_governance_domain_is_read_only_and_live(
assert row_count_after == row_count_before
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_hierarchical_lakehouse_domain_is_read_only_and_live(
+ transport: str,
+ doris_variant_sandbox: DorisVariantSandbox,
+) -> None:
+ environment = _server_environment(
+ doris_variant_sandbox.settings,
+ user=doris_variant_sandbox.settings.user,
+ password=doris_variant_sandbox.settings.password,
+ )
+ environment["MCP_TOOL_EXPOSURE_MODE"] = "hierarchical"
+ with doris_variant_sandbox.admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"SELECT COUNT(*) FROM {doris_variant_sandbox.qualified_table}"
+ )
+ row_count_before = int(cursor.fetchone()[0])
+
+ async with _transport_client(
+ transport,
+ environment,
+ read_timeout_seconds=60,
+ ) as client:
+ lakehouse_result = await client.call_tool("doris_lakehouse", {})
+ assert lakehouse_result.is_error is False
+ assert isinstance(lakehouse_result.structured_content, dict)
+ manifest = lakehouse_result.structured_content
+ assert manifest["mode"] == "manifest"
+ assert manifest["domain"] == "doris_lakehouse"
+ children = {child["name"]: child for child in manifest["children"]}
+ assert tuple(children) == LAKEHOUSE_CHILD_NAMES
+ assert all(
+ child["availability"]["callable"]
+ for child in children.values()
+ )
+ manifest_version = manifest["manifest_version"]
+
+ variant = await _call_domain_child(
+ client,
+ domain="doris_lakehouse",
+ child_tool="inspect_variant_column",
+ arguments={
+ "database": doris_variant_sandbox.settings.database,
+ "table": doris_variant_sandbox.table,
+ "column": "payload",
+ "path": "$.profile.age",
+ "sample_rows": 10,
+ },
+ manifest_version=manifest_version,
+ )
+ assert variant["data"]["shape_sample"]["rows_observed"] == 2
+ assert variant["data"]["shape_sample"]["paths"] == [
+ {
+ "path": "$.profile.age",
+ "types": [{"type": "BIGINT", "rows": 2}],
+ "rows_observed": 2,
+ "presence_ratio": 1.0,
+ }
+ ]
+ assert (
+ variant["data"]["advanced_capabilities"][
+ "storage_v3_supported"
+ ]
+ is False
+ )
+ assert variant["metadata"]["sampled_values_returned"] is False
+ assert doris_variant_sandbox.marker not in json.dumps(
+ variant,
+ ensure_ascii=False,
+ )
+
+ internal_catalog = await client.call_tool(
+ "doris_lakehouse",
+ {
+ "child_tool": "inspect_external_catalog",
+ "arguments": {"catalog": "internal"},
+ "manifest_version": manifest_version,
+ },
+ )
+ assert internal_catalog.is_error is True
+ assert doris_variant_sandbox.marker not in json.dumps(
+ internal_catalog.model_dump(by_alias=True, mode="json"),
+ ensure_ascii=False,
+ )
+
+ with doris_variant_sandbox.admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"SELECT COUNT(*) FROM {doris_variant_sandbox.qualified_table}"
+ )
+ row_count_after = int(cursor.fetchone()[0])
+ assert row_count_after == row_count_before
+
+
@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/protocol/test_multiworker_config.py
b/test/protocol/test_multiworker_config.py
index 929d965..4641378 100644
--- a/test/protocol/test_multiworker_config.py
+++ b/test/protocol/test_multiworker_config.py
@@ -216,6 +216,71 @@ def
test_governance_runtime_controls_load_serialize_and_validate(
)
+def test_lakehouse_runtime_controls_load_serialize_and_validate(
+ monkeypatch,
+ tmp_path,
+) -> None:
+ monkeypatch.setenv("LAKEHOUSE_MAX_CATALOG_OBJECTS", "80")
+ monkeypatch.setenv("LAKEHOUSE_MAX_CATALOG_DATABASES", "30")
+ monkeypatch.setenv("LAKEHOUSE_MAX_SNAPSHOTS", "90")
+ monkeypatch.setenv("LAKEHOUSE_MAX_PARTITIONS", "150")
+ monkeypatch.setenv("LAKEHOUSE_MAX_VARIANT_SAMPLE_ROWS", "40")
+ monkeypatch.setenv("LAKEHOUSE_MAX_VARIANT_PATHS", "300")
+
+ configured = DorisConfig.from_env()
+
+ assert configured.to_dict()["lakehouse"] == {
+ "max_catalog_objects": 80,
+ "max_catalog_databases": 30,
+ "max_snapshots": 90,
+ "max_partitions": 150,
+ "max_variant_sample_rows": 40,
+ "max_variant_paths": 300,
+ }
+ assert configured.validate() == []
+
+ config_path = tmp_path / "doris-mcp.json"
+ config_path.write_text(
+ json.dumps(
+ {
+ "lakehouse": {
+ "max_catalog_objects": 75,
+ "max_catalog_databases": 25,
+ "max_snapshots": 85,
+ "max_partitions": 140,
+ "max_variant_sample_rows": 35,
+ "max_variant_paths": 250,
+ }
+ }
+ ),
+ encoding="utf-8",
+ )
+ from_file = DorisConfig.from_file(str(config_path))
+ assert from_file.lakehouse.max_catalog_objects == 75
+ assert from_file.lakehouse.max_catalog_databases == 25
+ assert from_file.lakehouse.max_snapshots == 85
+ assert from_file.lakehouse.max_partitions == 140
+ assert from_file.lakehouse.max_variant_sample_rows == 35
+ assert from_file.lakehouse.max_variant_paths == 250
+
+ from_file.lakehouse.max_catalog_objects = 0
+ from_file.lakehouse.max_catalog_databases = 101
+ from_file.lakehouse.max_snapshots = 501
+ from_file.lakehouse.max_partitions = 1001
+ from_file.lakehouse.max_variant_sample_rows = 501
+ from_file.lakehouse.max_variant_paths = 2001
+ errors = from_file.validate()
+ assert "Lakehouse catalog object limit must be in the range 1-500" in
errors
+ assert (
+ "Lakehouse catalog database limit must be in the range 1-100"
+ in errors
+ )
+ assert "Lakehouse snapshot limit must be in the range 1-500" in errors
+ assert "Lakehouse partition limit must be in the range 1-1000" in errors
+ assert "Lakehouse Variant sample limit must be in the range 1-500" in
errors
+ assert "Lakehouse Variant path limit must be in the range 1-2000" in errors
+
+
def
test_state_handle_secret_and_ttl_are_configurable_without_serializing_secret(
monkeypatch,
):
@@ -264,6 +329,17 @@ def
test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
config.capability.snapshot_ttl_seconds = 120
config.capability.probe_timeout_seconds = 7
config.capability.stale_grace_seconds = 480
+ config.governance.max_sample_ratio = 0.15
+ config.governance.max_audit_window_days = 45
+ config.governance.max_lineage_edges = 750
+ config.governance.lineage_store_table = "governance.lineage_events"
+ config.governance.lineage_recent_event_minutes = 180
+ config.lakehouse.max_catalog_objects = 80
+ config.lakehouse.max_catalog_databases = 30
+ config.lakehouse.max_snapshots = 90
+ config.lakehouse.max_partitions = 150
+ config.lakehouse.max_variant_sample_rows = 40
+ config.lakehouse.max_variant_paths = 300
config.mcp_state_handle_secret = "parent-shared-state-handle-secret-value"
config.mcp_state_handle_ttl_seconds = 45
@@ -305,6 +381,20 @@ def
test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
assert child_config.capability.snapshot_ttl_seconds == 120
assert child_config.capability.probe_timeout_seconds == 7
assert child_config.capability.stale_grace_seconds == 480
+ assert child_config.governance.max_sample_ratio == 0.15
+ assert child_config.governance.max_audit_window_days == 45
+ assert child_config.governance.max_lineage_edges == 750
+ assert (
+ child_config.governance.lineage_store_table
+ == "governance.lineage_events"
+ )
+ assert child_config.governance.lineage_recent_event_minutes == 180
+ assert child_config.lakehouse.max_catalog_objects == 80
+ assert child_config.lakehouse.max_catalog_databases == 30
+ assert child_config.lakehouse.max_snapshots == 90
+ assert child_config.lakehouse.max_partitions == 150
+ assert child_config.lakehouse.max_variant_sample_rows == 40
+ assert child_config.lakehouse.max_variant_paths == 300
assert (
child_config.mcp_state_handle_secret
== "parent-shared-state-handle-secret-value"
diff --git a/test/tools/test_capability_detector.py
b/test/tools/test_capability_detector.py
index dc4ea0b..f83ff71 100644
--- a/test/tools/test_capability_detector.py
+++ b/test/tools/test_capability_detector.py
@@ -214,6 +214,64 @@ 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_lakehouse_probes_derive_target_sensitive_advanced_facets() ->
None:
+ connection = _ProbeConnection()
+ manager = _ProbeConnectionManager(connection)
+ detector = DorisCapabilityDetector(manager) # type: ignore[arg-type]
+ base = await detector.detect_base(
+ None,
+ capability_generation=1,
+ provider_generation="provider.lakehouse",
+ )
+
+ lakehouse = await detector.detect_domain(base, "doris_lakehouse", None)
+
+ assert lakehouse.probed_domains == frozenset({"doris_lakehouse"})
+ for probe_id in (
+ "external_catalog_metadata_readable",
+ "lakehouse_table_metadata_readable",
+ "variant_column_type_readable",
+ ):
+ assert (
+ lakehouse.probe(probe_id).status
+ is CapabilityProbeStatus.SUPPORTED
+ )
+ for probe_id in (
+ "lakehouse_snapshot_features_readable",
+ "iceberg_deletion_vector",
+ "iceberg_row_lineage",
+ ):
+ evidence = lakehouse.probe(probe_id)
+ assert evidence.status is CapabilityProbeStatus.DEGRADED
+ assert (
+ evidence.reason_code
+ == "TARGET_LAKEHOUSE_FORMAT_REQUIRES_CALL_TIME_VALIDATION"
+ )
+ for probe_id in (
+ "variant_advanced_properties_readable",
+ "variant_sparse_sharding",
+ "variant_sparse_cache",
+ "variant_doc_mode",
+ "storage_v3",
+ ):
+ evidence = lakehouse.probe(probe_id)
+ assert evidence.status is CapabilityProbeStatus.DEGRADED
+ assert (
+ evidence.reason_code
+ == "TARGET_VARIANT_PROPERTIES_REQUIRE_CALL_TIME_VALIDATION"
+ )
+ assert "SHOW CATALOGS" in connection.statements
+ assert (
+ "SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME "
+ "FROM information_schema.tables LIMIT 1"
+ ) in connection.statements
+ assert (
+ "SELECT COLUMN_NAME, DATA_TYPE "
+ "FROM information_schema.columns LIMIT 1"
+ ) in connection.statements
+
+
@pytest.mark.asyncio
async def test_search_probes_use_visible_target_and_isolated_connections() ->
None:
connection = _ProbeConnection()
diff --git a/test/tools/test_capability_registry.py
b/test/tools/test_capability_registry.py
index cc8d098..31dbf55 100644
--- a/test/tools/test_capability_registry.py
+++ b/test/tools/test_capability_registry.py
@@ -263,6 +263,113 @@ def
test_compaction_prefers_native_tracker_and_uses_legacy_on_405() -> None:
)
+def test_lakehouse_prefers_4_1_variants_and_falls_back_on_4_0() -> None:
+ bound = _BoundHandlers(
+ "doris_lakehouse.inspect_lakehouse_table",
+ "doris_lakehouse.inspect_variant_column",
+ )
+ evaluator = CapabilityEvaluator(
+ matrix=DORIS_FEATURE_MATRIX,
+ bound_handlers=bound, # type: ignore[arg-type]
+ )
+ domain = DORIS_DOMAIN_CATALOG.resolve_domain("doris_lakehouse")
+ table_child = DORIS_DOMAIN_CATALOG.resolve_child(
+ "doris_lakehouse",
+ "inspect_lakehouse_table",
+ )
+ variant_child = DORIS_DOMAIN_CATALOG.resolve_child(
+ "doris_lakehouse",
+ "inspect_variant_column",
+ )
+ probes = {
+ "lakehouse_table_metadata_readable": CapabilityProbeEvidence(
+ probe_id="lakehouse_table_metadata_readable",
+ status=CapabilityProbeStatus.SUPPORTED,
+ reason_code="TEST_BASELINE_METADATA",
+ ),
+ "variant_column_type_readable": CapabilityProbeEvidence(
+ probe_id="variant_column_type_readable",
+ status=CapabilityProbeStatus.SUPPORTED,
+ reason_code="TEST_BASELINE_VARIANT",
+ ),
+ **{
+ probe_id: CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=CapabilityProbeStatus.DEGRADED,
+ reason_code="TEST_CALL_TIME_TARGET_VALIDATION",
+ )
+ for probe_id in (
+ "lakehouse_snapshot_features_readable",
+ "iceberg_deletion_vector",
+ "iceberg_row_lineage",
+ "variant_advanced_properties_readable",
+ "variant_sparse_sharding",
+ "variant_sparse_cache",
+ "variant_doc_mode",
+ "storage_v3",
+ )
+ },
+ }
+ providers = CapabilityProviderRegistry(
+ {
+ "external_catalog_provider": CapabilityProviderEvidence(
+ provider_id="external_catalog_provider",
+ status=CapabilityProbeStatus.SUPPORTED,
+ reason_code="PROVIDER_CONFIGURED",
+ )
+ }
+ ).snapshot()
+ snapshot_405 = _snapshot(probes=probes)
+ snapshot_410 = replace(
+ snapshot_405,
+ version_vector=DorisClusterVersionVector.from_comments(
+ master_fe="Doris version doris-4.1.0",
+ follower_fes=("Doris version doris-4.1.0",),
+ backends=("Doris version doris-4.1.0",),
+ ),
+ )
+
+ table_405 = evaluator.evaluate(
+ snapshot=snapshot_405,
+ providers=providers,
+ domain=domain,
+ child=table_child,
+ auth_context=None,
+ )
+ table_410 = evaluator.evaluate(
+ snapshot=snapshot_410,
+ providers=providers,
+ domain=domain,
+ child=table_child,
+ auth_context=None,
+ )
+ variant_405 = evaluator.evaluate(
+ snapshot=snapshot_405,
+ providers=providers,
+ domain=domain,
+ child=variant_child,
+ auth_context=None,
+ )
+ variant_410 = evaluator.evaluate(
+ snapshot=snapshot_410,
+ providers=providers,
+ domain=domain,
+ child=variant_child,
+ auth_context=None,
+ )
+
+ assert table_405.active_variant == "lakehouse_table_metadata"
+ assert table_405.status is AvailabilityStatus.AVAILABLE
+ assert table_410.active_variant == "lakehouse_lifecycle_4_1"
+ assert table_410.status is AvailabilityStatus.DEGRADED
+ assert table_410.callable is True
+ assert variant_405.active_variant == "variant_type"
+ assert variant_405.status is AvailabilityStatus.AVAILABLE
+ assert variant_410.active_variant == "variant_advanced_4_1"
+ assert variant_410.status is AvailabilityStatus.DEGRADED
+ assert variant_410.callable is True
+
+
def test_evaluator_normalizes_system_object_probe_evidence_for_manifest() ->
None:
evaluator = CapabilityEvaluator(
matrix=DORIS_FEATURE_MATRIX,
@@ -418,6 +525,32 @@ def
test_builtin_query_evidence_providers_are_ready_when_bound() -> None:
assert providers["query_evidence_provider"].reason_code ==
("PROVIDER_CONFIGURED")
+def test_external_catalog_provider_is_ready_only_when_lakehouse_is_bound() ->
None:
+ bound_registry = CapabilityProviderRegistry.from_runtime(
+ matrix=DORIS_FEATURE_MATRIX,
+ bound_handlers=_BoundHandlers( # type: ignore[arg-type]
+ "doris_lakehouse.inspect_external_catalog",
+ "doris_lakehouse.inspect_lakehouse_table",
+ ),
+ config=SimpleNamespace(adbc=SimpleNamespace(enabled=False)),
+ )
+ unbound_registry = CapabilityProviderRegistry.from_runtime(
+ matrix=DORIS_FEATURE_MATRIX,
+ bound_handlers=_BoundHandlers(), # type: ignore[arg-type]
+ config=SimpleNamespace(adbc=SimpleNamespace(enabled=False)),
+ )
+
+ bound = bound_registry.snapshot().providers["external_catalog_provider"]
+ unbound = unbound_registry.snapshot().providers[
+ "external_catalog_provider"
+ ]
+
+ assert bound.status is CapabilityProbeStatus.SUPPORTED
+ assert bound.reason_code == "PROVIDER_CONFIGURED"
+ assert unbound.status is CapabilityProbeStatus.MISCONFIGURED
+ assert unbound.reason_code == "PROVIDER_NOT_CONFIGURED"
+
+
class _MutableClock:
def __init__(self) -> None:
self.now = datetime(2026, 7, 31, tzinfo=UTC)
diff --git a/test/tools/test_domain_dispatcher.py
b/test/tools/test_domain_dispatcher.py
index aeafae1..c615f96 100644
--- a/test/tools/test_domain_dispatcher.py
+++ b/test/tools/test_domain_dispatcher.py
@@ -53,6 +53,7 @@ from doris_mcp_server.tools.domain_models import (
from doris_mcp_server.tools.tools_manager import DorisToolsManager
from doris_mcp_server.utils.config import DorisConfig
from doris_mcp_server.utils.governance_runtime import GovernanceRuntimeFailure
+from doris_mcp_server.utils.lakehouse_runtime import LakehouseRuntimeFailure
from doris_mcp_server.utils.query_runtime import QueryRuntimeFailure
from doris_mcp_server.utils.security import AuthContext
@@ -163,6 +164,18 @@ def test_governance_domain_binds_all_eight_children() ->
None:
)
+def test_lakehouse_domain_binds_all_three_children() -> None:
+ manager = _manager()
+ bound = BoundHandlerAvailabilityProvider(manager)
+ lakehouse = DORIS_DOMAIN_CATALOG.resolve_domain("doris_lakehouse")
+
+ assert len(lakehouse.children) == 3
+ assert all(
+ bound.is_bound(lakehouse.name, child.name)
+ for child in lakehouse.children
+ )
+
+
@pytest.mark.asyncio
async def test_governance_failures_keep_stable_reason_codes() -> None:
manager = _manager("doris_governance.analyze_columns")
@@ -192,6 +205,36 @@ async def
test_governance_failures_keep_stable_reason_codes() -> None:
assert result.error.details["status_code"] == 404
[email protected]
+async def test_lakehouse_failures_keep_stable_reason_codes() -> None:
+ manager = _manager("doris_lakehouse.inspect_lakehouse_table")
+ manager.lakehouse_runtime.inspect_lakehouse_table = AsyncMock(
+ side_effect=LakehouseRuntimeFailure(
+ "The requested external table is unavailable.",
+ reason_code="LAKEHOUSE_TABLE_NOT_FOUND",
+ status_code=404,
+ )
+ )
+
+ result = await manager.domain_dispatcher.call_domain(
+ "doris_lakehouse",
+ {
+ "child_tool": "inspect_lakehouse_table",
+ "arguments": {
+ "catalog": "ice_prod",
+ "database": "analytics",
+ "table": "missing",
+ },
+ },
+ None,
+ )
+
+ assert result.mode == "error"
+ assert result.error.code is DomainErrorCode.CHILD_ARGUMENTS_INVALID
+ assert result.error.details["reason_code"] == "LAKEHOUSE_TABLE_NOT_FOUND"
+ assert result.error.details["status_code"] == 404
+
+
@pytest.mark.asyncio
async def test_cluster_metric_filters_reach_strict_runtime_unchanged() -> None:
manager = _manager("doris_cluster.get_monitoring_metrics")
diff --git a/test/tools/test_doris_feature_matrix.py
b/test/tools/test_doris_feature_matrix.py
index edf4ae6..49e48d2 100644
--- a/test/tools/test_doris_feature_matrix.py
+++ b/test/tools/test_doris_feature_matrix.py
@@ -138,6 +138,51 @@ def
test_search_prefers_vector_hybrid_and_keeps_text_fallback() -> None:
assert text.callable_when_degraded is True
+def test_lakehouse_prefers_4_1_capabilities_and_keeps_baseline_fallbacks() ->
None:
+ table_feature = DORIS_FEATURE_MATRIX.get_feature(
+ "doris_lakehouse",
+ "inspect_lakehouse_table",
+ )
+ variant_feature = DORIS_FEATURE_MATRIX.get_feature(
+ "doris_lakehouse",
+ "inspect_variant_column",
+ )
+
+ assert tuple(
+ variant.name for variant in table_feature.support_contract.variants
+ ) == (
+ "lakehouse_lifecycle_4_1",
+ "lakehouse_table_metadata",
+ )
+ table_advanced, table_baseline = table_feature.support_contract.variants
+ assert table_advanced.supported_ranges == (">=4.1.0",)
+ assert table_advanced.required_features == (
+ "iceberg_deletion_vector",
+ "iceberg_row_lineage",
+ )
+ assert table_advanced.callable_when_degraded is True
+ assert table_baseline.supported_ranges == (">=3.0.0",)
+
+ assert tuple(
+ variant.name for variant in variant_feature.support_contract.variants
+ ) == (
+ "variant_advanced_4_1",
+ "variant_type",
+ )
+ variant_advanced, variant_baseline = (
+ variant_feature.support_contract.variants
+ )
+ assert variant_advanced.supported_ranges == (">=4.1.0",)
+ assert variant_advanced.required_features == (
+ "variant_sparse_sharding",
+ "variant_sparse_cache",
+ "variant_doc_mode",
+ "storage_v3",
+ )
+ assert variant_advanced.callable_when_degraded is True
+ assert variant_baseline.supported_ranges == (">=3.0.0",)
+
+
def test_every_contract_is_fail_closed_and_has_resolvable_sources() -> None:
known_sources = {source.source_id for source in
DORIS_FEATURE_MATRIX.sources}
diff --git a/test/tools/test_lakehouse_handlers.py
b/test/tools/test_lakehouse_handlers.py
new file mode 100644
index 0000000..53f06f3
--- /dev/null
+++ b/test/tools/test_lakehouse_handlers.py
@@ -0,0 +1,163 @@
+# 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.
+
+"""Exact argument-routing contracts for the formal Lakehouse handlers."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from doris_mcp_server.tools.lakehouse_handlers import (
+ LakehouseToolHandlersMixin,
+)
+
+
+class _Harness(LakehouseToolHandlersMixin):
+ def __init__(self) -> None:
+ self.lakehouse_runtime = SimpleNamespace(
+ inspect_external_catalog=AsyncMock(
+ return_value={"status": "success"}
+ ),
+ inspect_lakehouse_table=AsyncMock(
+ return_value={"status": "success"}
+ ),
+ inspect_variant_column=AsyncMock(
+ return_value={"status": "success"}
+ ),
+ )
+
+
[email protected]
[email protected](
+ ("handler_name", "runtime_name", "arguments", "expected"),
+ [
+ (
+ "_formal_doris_lakehouse_inspect_external_catalog_tool",
+ "inspect_external_catalog",
+ {
+ "catalog": "ice_prod",
+ "include_objects": True,
+ "object_limit": 12,
+ },
+ {
+ "catalog": "ice_prod",
+ "include_objects": True,
+ "object_limit": 12,
+ },
+ ),
+ (
+ "_formal_doris_lakehouse_inspect_lakehouse_table_tool",
+ "inspect_lakehouse_table",
+ {
+ "catalog": "ice_prod",
+ "database": "analytics",
+ "table": "events",
+ "include_snapshots": True,
+ "include_partitions": True,
+ },
+ {
+ "catalog": "ice_prod",
+ "database": "analytics",
+ "table": "events",
+ "include_snapshots": True,
+ "include_partitions": True,
+ },
+ ),
+ (
+ "_formal_doris_lakehouse_inspect_variant_column_tool",
+ "inspect_variant_column",
+ {
+ "catalog": "internal",
+ "database": "analytics",
+ "table": "profiles",
+ "column": "payload",
+ "path": "$.profile.age",
+ "sample_rows": 10,
+ },
+ {
+ "catalog": "internal",
+ "database": "analytics",
+ "table": "profiles",
+ "column": "payload",
+ "path": "$.profile.age",
+ "sample_rows": 10,
+ },
+ ),
+ ],
+)
+async def test_lakehouse_handlers_forward_exact_arguments(
+ handler_name: str,
+ runtime_name: str,
+ arguments: dict[str, Any],
+ expected: dict[str, Any],
+) -> None:
+ harness = _Harness()
+
+ result = await getattr(harness, handler_name)(arguments)
+
+ assert result == {"status": "success"}
+ getattr(harness.lakehouse_runtime, runtime_name).assert_awaited_once_with(
+ **expected
+ )
+
+
[email protected]
+async def test_lakehouse_handlers_apply_documented_defaults() -> None:
+ harness = _Harness()
+
+ await harness._formal_doris_lakehouse_inspect_external_catalog_tool(
+ {"catalog": "ice_prod"}
+ )
+ await harness._formal_doris_lakehouse_inspect_lakehouse_table_tool(
+ {
+ "catalog": "ice_prod",
+ "database": "analytics",
+ "table": "events",
+ }
+ )
+ await harness._formal_doris_lakehouse_inspect_variant_column_tool(
+ {
+ "database": "analytics",
+ "table": "profiles",
+ "column": "payload",
+ }
+ )
+
+
harness.lakehouse_runtime.inspect_external_catalog.assert_awaited_once_with(
+ catalog="ice_prod",
+ include_objects=False,
+ object_limit=None,
+ )
+ harness.lakehouse_runtime.inspect_lakehouse_table.assert_awaited_once_with(
+ catalog="ice_prod",
+ database="analytics",
+ table="events",
+ include_snapshots=False,
+ include_partitions=False,
+ )
+ harness.lakehouse_runtime.inspect_variant_column.assert_awaited_once_with(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ path=None,
+ sample_rows=None,
+ )
diff --git a/test/utils/test_lakehouse_runtime.py
b/test/utils/test_lakehouse_runtime.py
new file mode 100644
index 0000000..1656c31
--- /dev/null
+++ b/test/utils/test_lakehouse_runtime.py
@@ -0,0 +1,519 @@
+# 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 Lakehouse runtime."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Mapping
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from doris_mcp_server.utils.lakehouse_runtime import (
+ DorisLakehouseRuntime,
+ LakehouseRuntimeFailure,
+)
+
+_Params = Mapping[str, Any] | tuple[Any, ...] | None
+_Responder = Callable[[str, _Params], list[dict[str, Any]]]
+
+
+class _ConnectionManager:
+ def __init__(
+ self,
+ responder: _Responder,
+ *,
+ lakehouse: Any | None = None,
+ ) -> None:
+ self._responder = responder
+ self.calls: list[str] = []
+ self.params: list[_Params] = []
+ self.max_rows: list[int] = []
+ self.context_count = 0
+ self.config = SimpleNamespace(
+ lakehouse=lakehouse
+ or SimpleNamespace(
+ max_catalog_objects=50,
+ max_catalog_databases=20,
+ max_snapshots=50,
+ max_partitions=100,
+ max_variant_sample_rows=20,
+ max_variant_paths=200,
+ )
+ )
+
+ @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: _Params = None,
+ **kwargs: Any,
+ ) -> SimpleNamespace:
+ manager.calls.append(sql)
+ manager.params.append(params)
+ manager.max_rows.append(int(kwargs["max_rows"]))
+ return SimpleNamespace(data=manager._responder(sql, params))
+
+ yield _Connection()
+
+
+def _runtime(
+ responder: _Responder,
+ *,
+ lakehouse: Any | None = None,
+) -> tuple[DorisLakehouseRuntime, _ConnectionManager]:
+ manager = _ConnectionManager(responder, lakehouse=lakehouse)
+ runtime = DorisLakehouseRuntime(manager) # type: ignore[arg-type]
+ return runtime, manager
+
+
+def _catalog_rows() -> list[dict[str, Any]]:
+ return [
+ {
+ "CatalogName": "ice_prod",
+ "Type": "iceberg",
+ "IsCurrent": "No",
+ "CreateTime": "2026-07-01 00:00:00",
+ "LastUpdateTime": "2026-07-30 10:00:00",
+ "Comment": "Warehouse endpoint https://comment.private.invalid";,
+ },
+ {
+ "CatalogName": "internal",
+ "Type": "internal",
+ "IsCurrent": "Yes",
+ },
+ ]
+
+
[email protected]
+async def test_external_catalog_returns_only_sanitized_bounded_metadata() ->
None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW CATALOGS":
+ return _catalog_rows()
+ if sql == "SHOW CATALOG `ice_prod`":
+ return [
+ {"Key": "type", "Value": "iceberg"},
+ {"Key": "password", "Value": "do-not-return"},
+ {
+ "Key": "s3.endpoint",
+ "Value": "https://private.example.invalid";,
+ },
+ {"Key": "use_meta_cache", "Value": "true"},
+ {"Key": "metadata_refresh_interval_sec", "Value": "60"},
+ {"Key": "warehouse", "Value": "s3://private/warehouse"},
+ ]
+ if sql == "SHOW DATABASES FROM `ice_prod`":
+ return [{"Database": "analytics"}, {"Database": "archive"}]
+ if sql == "SHOW FULL TABLES FROM `ice_prod`.`analytics`":
+ return [
+ {
+ "Tables_in_analytics": "events",
+ "Table_type": "BASE TABLE",
+ },
+ {
+ "Tables_in_analytics": "events_view",
+ "Table_type": "VIEW",
+ },
+ ]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, manager = _runtime(responder)
+ result = await runtime.inspect_external_catalog(
+ catalog="ice_prod",
+ include_objects=True,
+ object_limit=1,
+ )
+
+ assert result["status"] == "success"
+ assert result["data"]["reported_type"] == "iceberg"
+ assert result["data"]["configuration"] == {
+ "property_count": 6,
+ "property_keys": [
+ "metadata_refresh_interval_sec",
+ "type",
+ "use_meta_cache",
+ "warehouse",
+ ],
+ "property_keys_truncated": False,
+ "sensitive_property_count": 2,
+ "property_values_returned": False,
+ "metadata_cache_configured": True,
+ "refresh_interval_configured": True,
+ "warehouse_configured": True,
+ }
+ assert result["data"]["object_sample"]["relations"] == [
+ {"database": "analytics", "name": "events", "type": "table"}
+ ]
+ assert result["data"]["object_sample"]["truncated"] is True
+ serialized = str(result)
+ assert "do-not-return" not in serialized
+ assert "private.example.invalid" not in serialized
+ assert "comment.private.invalid" not in serialized
+ assert "s3://private/warehouse" not in serialized
+ assert manager.max_rows[-1] == 2
+
+
[email protected]
+async def test_external_catalog_rejects_internal_and_injected_names() -> None:
+ runtime, manager = _runtime(
+ lambda sql, _params: _catalog_rows() if sql == "SHOW CATALOGS" else []
+ )
+
+ with pytest.raises(LakehouseRuntimeFailure) as internal_failure:
+ await runtime.inspect_external_catalog(catalog="internal")
+ assert internal_failure.value.reason_code ==
"LAKEHOUSE_CATALOG_NOT_EXTERNAL"
+
+ call_count = len(manager.calls)
+ with pytest.raises(LakehouseRuntimeFailure) as injection_failure:
+ await runtime.inspect_external_catalog(catalog="ice; DROP DATABASE
prod")
+ assert injection_failure.value.reason_code == "LAKEHOUSE_ARGUMENT_INVALID"
+ assert len(manager.calls) == call_count
+
+
[email protected]
+async def test_external_catalog_degrades_when_properties_are_not_visible() ->
None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW CATALOGS":
+ return _catalog_rows()
+ if sql == "SHOW CATALOG `ice_prod`":
+ raise RuntimeError(1142, "catalog properties denied")
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, _ = _runtime(responder)
+ result = await runtime.inspect_external_catalog(catalog="ice_prod")
+
+ assert result["status"] == "partial"
+ assert result["data"]["configuration"]["property_count"] == 0
+ assert result["evidence"][1]["reason_code"] ==
"LAKEHOUSE_PERMISSION_DENIED"
+ assert "catalog properties denied" not in str(result)
+
+
[email protected]
+async def test_iceberg_table_returns_recorded_facets_without_raw_artifacts()
-> None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW CATALOGS":
+ return _catalog_rows()
+ if sql == "SHOW CATALOG `ice_prod`":
+ return [{"Key": "type", "Value": "iceberg"}]
+ if sql == "SHOW FULL COLUMNS FROM `ice_prod`.`analytics`.`events`":
+ return [
+ {
+ "Field": "event_id",
+ "Type": "BIGINT",
+ "Null": "NO",
+ "Key": "",
+ "Comment": "Stable identifier",
+ },
+ {
+ "Field": "_row_id",
+ "Type": "BIGINT",
+ "Null": "YES",
+ "Key": "",
+ "Comment": "",
+ },
+ ]
+ if sql == "SHOW CREATE TABLE `ice_prod`.`analytics`.`events`":
+ return [
+ {
+ "Create Table": (
+ "CREATE TABLE events (event_id BIGINT) "
+ "PARTITIONED BY (`event_date`) "
+ "PROPERTIES ('warehouse'='s3://private/location')"
+ )
+ }
+ ]
+ if sql == "SHOW TABLE STATS `ice_prod`.`analytics`.`events`":
+ return [
+ {
+ "row_count": "1000",
+ "data_size": "4096",
+ "update_time": "2026-07-30 12:00:00",
+ }
+ ]
+ if sql.startswith(
+ "SELECT * FROM `ice_prod`.`analytics`.`events$snapshots`"
+ ):
+ return [
+ {
+ "snapshot_id": 42,
+ "parent_id": 41,
+ "schema_id": 7,
+ "committed_at": "2026-07-30 12:00:00",
+ "operation": "append",
+ "record_count": 1000,
+ "summary": '{"private-location":"s3://secret"}',
+ }
+ ]
+ if sql.startswith(
+ "SHOW PARTITIONS FROM `ice_prod`.`analytics`.`events`"
+ ):
+ return [
+ {
+ "PartitionName": "event_date=2026-07-30",
+ "PartitionId": 9,
+ "Rows": 1000,
+ "UpdateTime": "2026-07-30 12:00:00",
+ }
+ ]
+ if sql.startswith(
+ "EXPLAIN SELECT * FROM `ice_prod`.`analytics`.`events`"
+ ):
+ return [
+ {
+ "Explain String": (
+ "ICEBERG_SCAN_NODE\nPARTITIONS=1/12\nPREDICATES:
event_id"
+ )
+ }
+ ]
+ if sql == "SELECT @@version_comment;":
+ return [
+ {
+ "@@version_comment": (
+ "Doris version doris-4.1.0-43f06a5e26 "
+ "(Cloud Mode)"
+ )
+ }
+ ]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, _ = _runtime(responder)
+ result = await runtime.inspect_lakehouse_table(
+ catalog="ice_prod",
+ database="analytics",
+ table="events",
+ include_snapshots=True,
+ include_partitions=True,
+ )
+
+ assert result["status"] == "success"
+ assert result["data"]["format"] == "iceberg"
+ assert result["data"]["partition_columns"] == ["event_date"]
+ assert result["data"]["statistics"]["row_count"] == 1000
+ assert result["data"]["snapshots"]["items"][0] == {
+ "snapshot_id": 42,
+ "parent_id": 41,
+ "schema_id": 7,
+ "committed_at": "2026-07-30 12:00:00",
+ "operation": "append",
+ "record_count": 1000,
+ "summary_available": True,
+ }
+ assert result["data"]["partitions"]["items"][0]["record_count"] == 1000
+ assert result["data"]["pushdown"] == {
+ "external_scan_observed": True,
+ "partition_pruning_observed": True,
+ "predicate_pushdown_observed": True,
+ "scan_nodes": ["ICEBERG_SCAN_NODE"],
+ }
+ assert result["data"]["lifecycle"]["iceberg_v3_lifecycle_eligible"] is True
+ assert result["data"]["lifecycle"]["row_lineage"][
+ "observable_hidden_columns"
+ ] == ["_row_id"]
+ serialized = str(result)
+ assert "CREATE TABLE events (" not in serialized
+ assert "s3://private/location" not in serialized
+ assert "ICEBERG_SCAN_NODE\nPARTITIONS" not in serialized
+ assert "s3://secret" not in serialized
+
+
[email protected]
+async def test_lakehouse_table_rejects_non_lakehouse_catalog() -> None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW CATALOGS":
+ return [{"CatalogName": "hms_prod", "Type": "hms"}]
+ if sql == "SHOW CATALOG `hms_prod`":
+ return [{"Key": "type", "Value": "hms"}]
+ if sql == "SHOW FULL COLUMNS FROM `hms_prod`.`analytics`.`events`":
+ return [{"Field": "id", "Type": "BIGINT", "Null": "NO"}]
+ if sql == "SHOW CREATE TABLE `hms_prod`.`analytics`.`events`":
+ return [{"Create Table": "CREATE TABLE events (id BIGINT)"}]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, _ = _runtime(responder)
+
+ with pytest.raises(LakehouseRuntimeFailure) as failure:
+ await runtime.inspect_lakehouse_table(
+ catalog="hms_prod",
+ database="analytics",
+ table="events",
+ )
+ assert (
+ failure.value.reason_code
+ == "LAKEHOUSE_TABLE_FORMAT_UNSUPPORTED"
+ )
+
+
[email protected]
+async def test_variant_inspection_binds_path_and_returns_only_type_shape() ->
None:
+ def responder(sql: str, params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW FULL COLUMNS FROM `analytics`.`profiles`":
+ return [
+ {
+ "Field": "payload",
+ "Type": "VARIANT<'$.profile.age': INT>",
+ "Null": "YES",
+ }
+ ]
+ if sql == "SHOW CREATE TABLE `analytics`.`profiles`":
+ return [
+ {
+ "Create Table": (
+ "CREATE TABLE profiles (payload VARIANT) "
+ "PROPERTIES ("
+ "'storage_format'='V3',"
+ "'variant_enable_doc_mode'='true',"
+ "'variant_sparse_hash_shard_count'='4',"
+ "'password'='do-not-return')"
+ )
+ }
+ ]
+ if sql.startswith(
+ "SELECT VARIANT_TYPE(`payload`[%s][%s]) AS `variant_type`"
+ ):
+ assert params == ("profile", "age")
+ assert sql.endswith("LIMIT 2")
+ return [
+ {"variant_type": '{"":"int"}'},
+ {"variant_type": '{"":"bigint"}'},
+ ]
+ if sql == "SELECT @@version_comment;":
+ return [{"@@version_comment": "Apache Doris 4.1.0"}]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, manager = _runtime(responder)
+ result = await runtime.inspect_variant_column(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ path="$.profile.age",
+ sample_rows=2,
+ )
+
+ assert result["status"] == "partial"
+ assert result["data"]["typed_paths"] == [
+ {"path": "$.profile.age", "type": "INT"}
+ ]
+ assert result["data"]["configuration"]["mode"] == "doc"
+ assert result["data"]["configuration"]["storage_v3"] is True
+ assert result["data"]["advanced_capabilities"]["doc_mode_supported"] is
True
+ assert result["data"]["shape_sample"]["paths"] == [
+ {
+ "path": "$.profile.age",
+ "types": [
+ {"type": "INT", "rows": 1},
+ {"type": "BIGINT", "rows": 1},
+ ],
+ "rows_observed": 2,
+ "presence_ratio": 1.0,
+ }
+ ]
+ assert manager.params[-2] == ("profile", "age")
+ serialized = str(result)
+ assert "do-not-return" not in serialized
+ assert "CREATE TABLE profiles (" not in serialized
+ assert result["metadata"]["sampled_values_returned"] is False
+
+
[email protected]
+async def test_variant_sampling_failure_degrades_without_exposing_values() ->
None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW FULL COLUMNS FROM `analytics`.`profiles`":
+ return [{"Field": "payload", "Type": "VARIANT", "Null": "YES"}]
+ if sql == "SHOW CREATE TABLE `analytics`.`profiles`":
+ return [{"Create Table": "CREATE TABLE profiles (payload
VARIANT)"}]
+ if sql.startswith("SELECT VARIANT_TYPE"):
+ raise RuntimeError(1142, "SELECT denied for secret principal")
+ if sql == "SELECT @@version_comment;":
+ return [{"@@version_comment": "Apache Doris 4.0.6"}]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ runtime, _ = _runtime(responder)
+ result = await runtime.inspect_variant_column(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ )
+
+ assert result["status"] == "partial"
+ assert result["data"]["shape_sample"]["rows_observed"] == 0
+ assert result["data"]["shape_sample"]["paths"] == []
+ assert result["evidence"][-1]["reason_code"] ==
"LAKEHOUSE_PERMISSION_DENIED"
+ assert "secret principal" not in str(result)
+
+
[email protected]
+async def
test_variant_rejects_non_variant_path_injection_and_oversized_sample() -> None:
+ def responder(sql: str, _params: _Params) -> list[dict[str, Any]]:
+ if sql == "SHOW FULL COLUMNS FROM `analytics`.`profiles`":
+ return [{"Field": "payload", "Type": "VARCHAR", "Null": "YES"}]
+ raise AssertionError(f"Unexpected SQL: {sql}")
+
+ lakehouse = SimpleNamespace(
+ max_catalog_objects=50,
+ max_catalog_databases=20,
+ max_snapshots=50,
+ max_partitions=100,
+ max_variant_sample_rows=5,
+ max_variant_paths=200,
+ )
+ runtime, manager = _runtime(responder, lakehouse=lakehouse)
+
+ with pytest.raises(LakehouseRuntimeFailure) as type_failure:
+ await runtime.inspect_variant_column(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ )
+ assert type_failure.value.reason_code == "LAKEHOUSE_COLUMN_NOT_VARIANT"
+
+ call_count = len(manager.calls)
+ with pytest.raises(LakehouseRuntimeFailure) as path_failure:
+ await runtime.inspect_variant_column(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ path="$['profile']; DROP TABLE users",
+ )
+ assert path_failure.value.reason_code == "LAKEHOUSE_ARGUMENT_INVALID"
+ assert len(manager.calls) == call_count
+
+ with pytest.raises(LakehouseRuntimeFailure) as limit_failure:
+ await runtime.inspect_variant_column(
+ catalog=None,
+ database="analytics",
+ table="profiles",
+ column="payload",
+ sample_rows=6,
+ )
+ assert limit_failure.value.reason_code == "LAKEHOUSE_ARGUMENT_INVALID"
+ assert len(manager.calls) == call_count
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
