+------------------------------------------------------------------+
|                        SearchService                              |
|                                                                   |
|  +-----------+  +-----------+  +-----------+  +----------------+  |
|  |  Query    |  |  Facet    |  | Suggest   |  |  Saved Search  |  |
|  |  Engine   |  |  Engine   |  | Engine    |  |  Manager       |  |
|  +-----------+  +-----------+  +-----------+  +----------------+  |
|       |              |              |                |             |
|  +-----------+  +-----------+  +-----------+  +----------------+  |
|  |  Mode     |  |  Agg      |  |  Redis    |  |  PostgreSQL    |  |
|  |  Router   |  |  Builder  |  |  Hot/Hist |  |  Storage       |  |
|  +-----------+  +-----------+  +-----------+  +----------------+  |
|       |              |                                            |
|  +----+--------------+----------------------------------------+   |
|  |            Permission-Aware Filter Layer                   |   |
|  +------------------------------------------------------------+   |
|       |                                                           |
|  +------------------------------------------------------------+   |
|  |          Doris Inverted Index / Full-Text Engine            |   |
|  +------------------------------------------------------------+   |
+------------------------------------------------------------------+

message SearchRequest {
  string world_id = 1;
  string query = 2;
  SearchMode mode = 3;
  repeated string object_types = 4;    // restrict search to types
  repeated string search_fields = 5;   // restrict search to fields
  repeated FacetRequest facets = 6;
  repeated FilterClause filters = 7;
  Pagination pagination = 8;
  SortSpec sort = 9;
  bool include_highlights = 10;
  bool include_relations = 11;
}

enum SearchMode {
  BEST_MATCH = 0;  // default: composite scoring
  PREFIX = 1;       // prefix matching
  FUZZY = 2;        // fuzzy matching (tolerant)
  EXACT = 3;        // exact matching
  WILDCARD = 4;     // wildcard patterns
  REGEX = 5;        // regular expressions
}

message SearchResponse {
  repeated SearchHit hits = 1;
  int64 total_count = 2;
  repeated FacetResult facets = 3;
  repeated Suggestion suggestions = 4;
  SearchMetadata metadata = 5;
}

message SearchHit {
  string object_type = 1;
  string object_id = 2;
  double score = 3;
  map<string, Value> attributes = 4;
  repeated Highlight highlights = 5;
  repeated RelatedObject relations = 6;
}

Score = w1 * text_relevance
      + w2 * field_boost
      + w3 * recency_score
      + w4 * popularity_score
      + w5 * type_priority

class BestMatchScorer:
    """Composite scorer for BEST_MATCH mode"""

    FIELD_WEIGHTS = {
        "title": 3.0,
        "display_name": 3.0,
        "description": 1.5,
        "tags": 2.0,
        "properties": 1.0,
    }

    def score(self, query: str, hit: RawHit) -> float:
        text_score = hit.bm25_score
        field_score = self.FIELD_WEIGHTS.get(hit.matched_field, 1.0)
        recency = self._recency_decay(hit.updated_at)
        popularity = math.log1p(hit.view_count) * 0.1
        type_priority = self._type_priority(hit.object_type)

        return (
            0.5 * text_score * field_score
            + 0.15 * recency
            + 0.1 * popularity
            + 0.25 * type_priority
        )

    def _recency_decay(self, updated_at: datetime) -> float:
        days_ago = (datetime.utcnow() - updated_at).days
        return math.exp(-0.01 * days_ago)  # half-life ~70 days

    def _type_priority(self, object_type: str) -> float:
        priorities = {"Customer": 1.0, "Order": 0.9, "Product": 0.85}
        return priorities.get(object_type, 0.5)

-- Doris prefix query
SELECT * FROM ontology_objects
WHERE display_name LIKE 'cust%'
ORDER BY length(display_name) ASC, view_count DESC
LIMIT 10;

CREATE INDEX idx_display_name ON ontology_objects(display_name)
  USING INVERTED PROPERTIES("parser" = "standard");

class FuzzyMatcher:
    """Edit-distance-based fuzzy matching"""

    def __init__(self, max_edit_distance: int = 2):
        self.max_edit_distance = max_edit_distance

    def match(self, query: str, candidates: list[str]) -> list[tuple[str, int]]:
        results = []
        for candidate in candidates:
            distance = self._levenshtein(query.lower(), candidate.lower())
            if distance <= self.max_edit_distance:
                results.append((candidate, distance))
        return sorted(results, key=lambda x: x[1])

    def _levenshtein(self, s1: str, s2: str) -> int:
        if len(s1) < len(s2):
            return self._levenshtein(s2, s1)
        if len(s2) == 0:
            return len(s1)
        prev_row = range(len(s2) + 1)
        for i, c1 in enumerate(s1):
            curr_row = [i + 1]
            for j, c2 in enumerate(s2):
                insertions = prev_row[j + 1] + 1
                deletions = curr_row[j] + 1
                substitutions = prev_row[j] + (c1 != c2)
                curr_row.append(min(insertions, deletions, substitutions))
            prev_row = curr_row
        return prev_row[-1]

SELECT * FROM ontology_objects
WHERE object_id = 'ORD-2024-001234'
   OR attributes->>'sku' = 'ORD-2024-001234';

class WildcardMatcher:
    """Wildcard search converter"""

    def to_sql(self, pattern: str) -> str:
        """Convert wildcard pattern to SQL LIKE pattern"""
        sql_pattern = pattern.replace("*", "%").replace("?", "_")
        return f"display_name LIKE '{sql_pattern}'"

    def to_regex(self, pattern: str) -> str:
        """Convert wildcard pattern to regex"""
        regex = pattern.replace(".", r"\.") \
                       .replace("*", ".*") \
                       .replace("?", ".")
        return f"^{regex}$"

class RegexSearchHandler:
    """Regular expression search handler"""

    MAX_REGEX_LENGTH = 200
    TIMEOUT_SECONDS = 5

    DANGEROUS_PATTERNS = [
        r"(.+)+",     # exponential backtracking
        r"(a*)*",     # nested quantifiers
        r"(a|a)*",    # overlapping alternation
    ]

    def validate(self, pattern: str) -> None:
        if len(pattern) > self.MAX_REGEX_LENGTH:
            raise SearchError("Regex pattern too long")
        for dangerous in self.DANGEROUS_PATTERNS:
            if dangerous in pattern:
                raise SearchError("Potentially catastrophic regex pattern")
        try:
            re.compile(pattern)
        except re.error as e:
            raise SearchError(f"Invalid regex: {e}")

    def search(self, pattern: str, field: str) -> str:
        self.validate(pattern)
        return f"{field} REGEXP '{pattern}'"

class SearchModeRouter:
    """Routes to the appropriate handler based on search mode"""

    def __init__(self):
        self._handlers: dict[SearchMode, SearchHandler] = {
            SearchMode.BEST_MATCH: BestMatchHandler(),
            SearchMode.PREFIX: PrefixHandler(),
            SearchMode.FUZZY: FuzzyHandler(),
            SearchMode.EXACT: ExactHandler(),
            SearchMode.WILDCARD: WildcardHandler(),
            SearchMode.REGEX: RegexHandler(),
        }

    def route(self, request: SearchRequest) -> SearchResponse:
        handler = self._handlers.get(request.mode)
        if handler is None:
            raise SearchError(f"Unsupported search mode: {request.mode}")

        # Auto-detect mode (when mode=BEST_MATCH)
        if request.mode == SearchMode.BEST_MATCH:
            detected = self._auto_detect(request.query)
            if detected != SearchMode.BEST_MATCH:
                handler = self._handlers[detected]

        return handler.execute(request)

    def _auto_detect(self, query: str) -> SearchMode:
        if query.startswith('"') and query.endswith('"'):
            return SearchMode.EXACT
        if "*" in query or "?" in query:
            return SearchMode.WILDCARD
        if query.startswith("/") and query.endswith("/"):
            return SearchMode.REGEX
        return SearchMode.BEST_MATCH

message FacetRequest {
  string field = 1;
  FacetType type = 2;
  int32 size = 3;            // TERMS: return top N values
  repeated double ranges = 4; // RANGE: interval boundaries
  repeated DateRange date_ranges = 5;
}

enum FacetType {
  TERMS = 0;
  RANGE = 1;
  DATE_RANGE = 2;
}

message FacetResult {
  string field = 1;
  FacetType type = 2;
  repeated FacetBucket buckets = 3;
}

message FacetBucket {
  string key = 1;
  int64 count = 2;
  double from = 3;   // RANGE
  double to = 4;     // RANGE
}

class FacetQueryBuilder:
    """Facet query builder"""

    def build_terms_facet(self, field: str, size: int) -> str:
        return f"""
        SELECT {field} AS facet_key, COUNT(*) AS facet_count
        FROM ontology_objects
        WHERE {{base_where_clause}}
        GROUP BY {field}
        ORDER BY facet_count DESC
        LIMIT {size}
        """

    def build_range_facet(self, field: str, ranges: list[float]) -> str:
        cases = []
        for i in range(len(ranges) - 1):
            lo, hi = ranges[i], ranges[i + 1]
            cases.append(
                f"WHEN {field} >= {lo} AND {field} < {hi} "
                f"THEN '{lo}-{hi}'"
            )
        cases.append(f"WHEN {field} >= {ranges[-1]} THEN '{ranges[-1]}+'")

        case_sql = "\n            ".join(cases)
        return f"""
        SELECT
            CASE
                {case_sql}
            END AS facet_key,
            COUNT(*) AS facet_count
        FROM ontology_objects
        WHERE {{base_where_clause}}
        GROUP BY facet_key
        ORDER BY facet_key
        """

    def build_date_range_facet(
        self, field: str, ranges: list[dict]
    ) -> str:
        cases = []
        for r in ranges:
            label = r["label"]
            from_date = r.get("from", "1970-01-01")
            to_date = r.get("to", "2099-12-31")
            cases.append(
                f"WHEN {field} >= '{from_date}' "
                f"AND {field} < '{to_date}' "
                f"THEN '{label}'"
            )
        case_sql = "\n            ".join(cases)
        return f"""
        SELECT
            CASE
                {case_sql}
            END AS facet_key,
            COUNT(*) AS facet_count
        FROM ontology_objects
        WHERE {{base_where_clause}}
        GROUP BY facet_key
        """

class FacetFilterCoordinator:
    """Coordinates facet filtering with facet statistics"""

    def execute_with_facets(
        self,
        base_query: str,
        facet_requests: list[FacetRequest],
        active_filters: dict[str, list[str]],
    ) -> tuple[list[SearchHit], list[FacetResult]]:
        # Main query applies all filters
        main_results = self._execute_main(base_query, active_filters)

        # Each facet query excludes its own filter
        facet_results = []
        for facet in facet_requests:
            other_filters = {
                k: v for k, v in active_filters.items()
                if k != facet.field
            }
            facet_result = self._execute_facet(
                base_query, facet, other_filters
            )
            facet_results.append(facet_result)

        return main_results, facet_results

+-------------------------------------------+
|  Layer 1: Hot Queries (Redis Sorted Set)  |  <-- global trending
+-------------------------------------------+
|  Layer 2: Recent History (per user)       |  <-- personal history
+-------------------------------------------+
|  Layer 3: Entity Name Index (Trie)        |  <-- entity names
+-------------------------------------------+

class HotQueryManager:
    """Hot query management using Redis Sorted Sets"""

    HOT_QUERY_KEY = "search:hot_queries"
    RECENT_KEY_PREFIX = "search:recent:{user_id}"
    MAX_HOT_QUERIES = 1000
    MAX_RECENT = 50
    HOT_QUERY_TTL = 86400 * 7  # 7 days

    def __init__(self, redis: Redis):
        self._redis = redis

    async def record_query(
        self, query: str, user_id: str
    ) -> None:
        """Record a search query"""
        normalized = query.strip().lower()
        if len(normalized) < 2:
            return

        pipe = self._redis.pipeline()
        # Increment hot query count
        pipe.zincrby(self.HOT_QUERY_KEY, 1, normalized)
        # Personal history (most recent 50)
        pipe.lpush(f"{self.RECENT_KEY_PREFIX}:{user_id}", normalized)
        pipe.ltrim(f"{self.RECENT_KEY_PREFIX}:{user_id}", 0, self.MAX_RECENT - 1)
        pipe.expire(f"{self.RECENT_KEY_PREFIX}:{user_id}", self.HOT_QUERY_TTL)
        await pipe.execute()

        # Periodically clean low-frequency terms
        count = await self._redis.zcard(self.HOT_QUERY_KEY)
        if count > self.MAX_HOT_QUERIES:
            await self._redis.zremrangebyrank(
                self.HOT_QUERY_KEY, 0,
                count - self.MAX_HOT_QUERIES - 1
            )

    async def get_suggestions(
        self, prefix: str, user_id: str, limit: int = 10
    ) -> list[Suggestion]:
        """Get search suggestions"""
        suggestions = []

        # 1. Personal history (priority)
        recent = await self._redis.lrange(
            f"{self.RECENT_KEY_PREFIX}:{user_id}", 0, -1
        )
        for q in recent:
            q_str = q.decode() if isinstance(q, bytes) else q
            if q_str.startswith(prefix.lower()):
                suggestions.append(
                    Suggestion(text=q_str, source="RECENT", score=1.0)
                )
            if len(suggestions) >= limit // 3:
                break

        # 2. Global hot queries
        hot = await self._redis.zrevrangebyscore(
            self.HOT_QUERY_KEY, "+inf", "-inf",
            start=0, num=100, withscores=True
        )
        for q, score in hot:
            q_str = q.decode() if isinstance(q, bytes) else q
            if q_str.startswith(prefix.lower()):
                suggestions.append(
                    Suggestion(text=q_str, source="HOT", score=score)
                )
            if len(suggestions) >= limit:
                break

        return suggestions[:limit]

class EntitySuggester:
    """Entity name-based search suggestions"""

    SUGGEST_SQL = """
    SELECT display_name, object_type, object_id
    FROM ontology_objects
    WHERE display_name LIKE '{prefix}%'
    ORDER BY view_count DESC
    LIMIT {limit}
    """

    async def suggest(
        self, prefix: str, limit: int = 5
    ) -> list[Suggestion]:
        rows = await self._doris.execute(
            self.SUGGEST_SQL.format(prefix=prefix, limit=limit)
        )
        return [
            Suggestion(
                text=row["display_name"],
                source="ENTITY",
                score=0.5,
                metadata={
                    "object_type": row["object_type"],
                    "object_id": row["object_id"],
                },
            )
            for row in rows
        ]

class SavedSearch(BaseModel):
    """A saved search configuration"""
    id: str = Field(default_factory=lambda: str(uuid4()))
    user_id: str
    name: str
    description: str | None = None
    query: str
    mode: SearchMode = SearchMode.BEST_MATCH
    object_types: list[str] = []
    filters: list[FilterClause] = []
    facet_selections: dict[str, list[str]] = {}
    sort: SortSpec | None = None
    is_shared: bool = False
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    last_executed_at: datetime | None = None
    execution_count: int = 0

class SavedSearchService:
    """Saved search management service"""

    async def create(
        self, user_id: str, request: CreateSavedSearchRequest
    ) -> SavedSearch:
        saved = SavedSearch(
            user_id=user_id,
            name=request.name,
            description=request.description,
            query=request.query,
            mode=request.mode,
            object_types=request.object_types,
            filters=request.filters,
        )
        await self._repo.save(saved)
        return saved

    async def execute(
        self, saved_id: str, user_id: str
    ) -> SearchResponse:
        saved = await self._repo.get(saved_id)
        if saved.user_id != user_id and not saved.is_shared:
            raise PermissionError("Cannot execute others' private search")

        request = SearchRequest(
            query=saved.query,
            mode=saved.mode,
            object_types=saved.object_types,
            filters=saved.filters,
        )

        saved.last_executed_at = datetime.utcnow()
        saved.execution_count += 1
        await self._repo.save(saved)

        return await self._search_service.search(request)

    async def list_by_user(
        self, user_id: str
    ) -> list[SavedSearch]:
        own = await self._repo.find_by_user(user_id)
        shared = await self._repo.find_shared()
        return own + [s for s in shared if s.user_id != user_id]

+--------------------------------------------------+
|  SearchRequest                                    |
|       |                                           |
|       v                                           |
|  Query Execution (Doris)                          |
|       |                                           |
|       v                                           |
|  Raw Results (N hits)                             |
|       |                                           |
|       v                                           |
|  Permission Filter Layer                          |
|  +----------------------------------------------+|
|  |  1. Object-type-level ACL                    ||
|  |  2. Row-level security (RLS)                 ||
|  |  3. Column-level masking                     ||
|  |  4. World-level isolation                    ||
|  +----------------------------------------------+|
|       |                                           |
|  Filtered Results (M hits, M <= N)                |
+--------------------------------------------------+

class PermissionAwareSearchFilter:
    """Permission-aware search filter"""

    async def inject_permission_clause(
        self, user: User, base_query: str
    ) -> str:
        # Get object types the user can access
        accessible_types = await self._acl_service.get_accessible_types(
            user.id, Permission.READ
        )
        type_clause = (
            f"object_type IN ({','.join(repr(t) for t in accessible_types)})"
        )

        # Get row-level security policies
        rls_policies = await self._policy_service.get_rls_policies(user.id)
        rls_clauses = []
        for policy in rls_policies:
            rls_clauses.append(policy.to_sql_clause())

        # Get World isolation condition
        world_clause = f"world_id = '{user.active_world_id}'"

        # Combine all permission conditions
        permission_clause = " AND ".join(
            [type_clause, world_clause] + rls_clauses
        )

        return f"{base_query} AND ({permission_clause})"

    async def mask_columns(
        self, user: User, hits: list[SearchHit]
    ) -> list[SearchHit]:
        """Apply column-level masking to search results"""
        masking_rules = await self._policy_service.get_masking_rules(user.id)

        for hit in hits:
            for field, rule in masking_rules.items():
                if field in hit.attributes:
                    hit.attributes[field] = rule.apply(
                        hit.attributes[field]
                    )
                # Highlights also need masking
                hit.highlights = [
                    h for h in hit.highlights
                    if h.field not in masking_rules
                ]

        return hits

-- Create inverted indexes on the objects table
ALTER TABLE ontology_objects ADD INDEX idx_inv_display_name(display_name)
  USING INVERTED
  PROPERTIES(
    "parser" = "chinese",
    "lower_case" = "true"
  );

ALTER TABLE ontology_objects ADD INDEX idx_inv_description(description)
  USING INVERTED
  PROPERTIES(
    "parser" = "chinese",
    "lower_case" = "true",
    "support_phrase" = "true"
  );

ALTER TABLE ontology_objects ADD INDEX idx_inv_tags(tags)
  USING INVERTED
  PROPERTIES(
    "parser" = "comma"
  );

-- Create inverted index on JSON attributes column
ALTER TABLE ontology_objects ADD INDEX idx_inv_attrs(attributes)
  USING INVERTED
  PROPERTIES(
    "parser" = "chinese",
    "lower_case" = "true"
  );

-- MATCH_ANY: any token matches
SELECT * FROM ontology_objects
WHERE display_name MATCH_ANY 'customer order';

-- MATCH_ALL: all tokens must match
SELECT * FROM ontology_objects
WHERE description MATCH_ALL 'data pipeline scheduler';

-- MATCH_PHRASE: phrase matching (preserves token order)
SELECT * FROM ontology_objects
WHERE description MATCH_PHRASE 'real-time data pipeline';

class DorisSearchQueryBuilder:
    """Doris full-text search query builder"""

    def build(self, request: SearchRequest) -> str:
        select_clause = self._build_select(request)
        where_clause = self._build_where(request)
        order_clause = self._build_order(request)
        limit_clause = self._build_limit(request.pagination)

        return f"""
        {select_clause}
        FROM ontology_objects
        WHERE {where_clause}
        {order_clause}
        {limit_clause}
        """

    def _build_where(self, request: SearchRequest) -> str:
        clauses = []

        # Search conditions
        if request.mode == SearchMode.BEST_MATCH:
            fields = request.search_fields or [
                "display_name", "description", "tags"
            ]
            match_clauses = [
                f"{f} MATCH_ANY '{request.query}'" for f in fields
            ]
            clauses.append(f"({' OR '.join(match_clauses)})")

        elif request.mode == SearchMode.PREFIX:
            fields = request.search_fields or ["display_name"]
            like_clauses = [
                f"{f} LIKE '{request.query}%'" for f in fields
            ]
            clauses.append(f"({' OR '.join(like_clauses)})")

        elif request.mode == SearchMode.EXACT:
            fields = request.search_fields or ["display_name"]
            exact_clauses = [
                f"{f} = '{request.query}'" for f in fields
            ]
            clauses.append(f"({' OR '.join(exact_clauses)})")

        # Type filter
        if request.object_types:
            types = ",".join(f"'{t}'" for t in request.object_types)
            clauses.append(f"object_type IN ({types})")

        # Additional filters
        for f in request.filters:
            clauses.append(self._filter_to_sql(f))

        return " AND ".join(clauses) if clauses else "1=1"

class HighlightGenerator:
    """Search result highlight generator"""

    PRE_TAG = "<em>"
    POST_TAG = "</em>"
    FRAGMENT_SIZE = 150

    def highlight(
        self, query: str, text: str, field: str
    ) -> list[Highlight]:
        if not text:
            return []

        tokens = self._tokenize(query)
        fragments = self._extract_fragments(text, tokens)

        return [
            Highlight(
                field=field,
                fragment=self._mark_tokens(fragment, tokens),
            )
            for fragment in fragments
        ]

    def _tokenize(self, query: str) -> list[str]:
        """Tokenize the query"""
        return [t.strip() for t in query.split() if t.strip()]

    def _extract_fragments(
        self, text: str, tokens: list[str]
    ) -> list[str]:
        """Extract text fragments containing matched tokens"""
        fragments = []
        text_lower = text.lower()

        for token in tokens:
            pos = text_lower.find(token.lower())
            if pos >= 0:
                start = max(0, pos - self.FRAGMENT_SIZE // 2)
                end = min(len(text), pos + len(token) + self.FRAGMENT_SIZE // 2)
                fragment = text[start:end]
                if start > 0:
                    fragment = "..." + fragment
                if end < len(text):
                    fragment = fragment + "..."
                fragments.append(fragment)

        return fragments[:3]  # max 3 fragments

    def _mark_tokens(self, text: str, tokens: list[str]) -> str:
        """Mark matched tokens in text"""
        result = text
        for token in tokens:
            pattern = re.compile(re.escape(token), re.IGNORECASE)
            result = pattern.sub(
                f"{self.PRE_TAG}\\g<0>{self.POST_TAG}", result
            )
        return result

class SearchCache:
    """Search result cache"""

    CACHE_TTL = 300  # 5 minutes
    CACHE_PREFIX = "search:cache:"

    async def get_or_execute(
        self, request: SearchRequest, executor: Callable
    ) -> SearchResponse:
        cache_key = self._build_key(request)
        cached = await self._redis.get(cache_key)

        if cached:
            return SearchResponse.model_validate_json(cached)

        response = await executor(request)

        # Only cache queries that take significant time
        if response.metadata.query_time_ms > 100:
            await self._redis.setex(
                cache_key,
                self.CACHE_TTL,
                response.model_dump_json(),
            )

        return response

    def _build_key(self, request: SearchRequest) -> str:
        import hashlib
        content = request.model_dump_json()
        digest = hashlib.sha256(content.encode()).hexdigest()[:16]
        return f"{self.CACHE_PREFIX}{digest}"

class SearchCircuitBreaker:
    """Search query circuit breaker"""

    def __init__(
        self,
        timeout_ms: int = 5000,
        failure_threshold: int = 5,
        reset_timeout: int = 30,
    ):
        self._timeout_ms = timeout_ms
        self._failure_count = 0
        self._failure_threshold = failure_threshold
        self._reset_timeout = reset_timeout
        self._state = "CLOSED"
        self._last_failure: datetime | None = None

    async def execute(
        self, coro: Coroutine
    ) -> SearchResponse:
        if self._state == "OPEN":
            if self._should_reset():
                self._state = "HALF_OPEN"
            else:
                raise SearchError("Search service circuit breaker is OPEN")

        try:
            result = await asyncio.wait_for(
                coro, timeout=self._timeout_ms / 1000
            )
            self._on_success()
            return result
        except asyncio.TimeoutError:
            self._on_failure()
            raise SearchError(f"Search timed out after {self._timeout_ms}ms")
        except Exception:
            self._on_failure()
            raise

service SearchService {
  // General search
  rpc Search(SearchRequest) returns (SearchResponse);

  // Search suggestions
  rpc Suggest(SuggestRequest) returns (SuggestResponse);

  // Facet statistics only
  rpc GetFacets(FacetOnlyRequest) returns (FacetOnlyResponse);

  // Saved searches
  rpc CreateSavedSearch(CreateSavedSearchRequest)
      returns (SavedSearch);
  rpc ListSavedSearches(ListSavedSearchesRequest)
      returns (ListSavedSearchesResponse);
  rpc ExecuteSavedSearch(ExecuteSavedSearchRequest)
      returns (SearchResponse);
  rpc DeleteSavedSearch(DeleteSavedSearchRequest)
      returns (google.protobuf.Empty);
}

User types "customer Zhang"
       |
       v
  AutoSuggest (< 50ms)
  +-- Recent: "customer Zhang San" (user's history)
  +-- Hot: "customer management" (global hot)
  +-- Entity: "Zhang Wei" (Customer), "Zhang Ming" (Supplier)
       |
  User selects or presses Enter
       |
       v
  SearchRequest(query="customer Zhang San", mode=BEST_MATCH)
       |
       v
  ModeRouter -> auto-detect -> BEST_MATCH
       |
       v
  PermissionFilter.inject(user) -> add ACL/RLS clauses
       |
       v
  DorisQueryBuilder.build() ->
    SELECT * FROM ontology_objects
    WHERE (display_name MATCH_ALL 'customer Zhang San'
           OR description MATCH_ANY 'customer Zhang San')
      AND object_type IN ('Customer','Order')
      AND world_id = 'main'
    ORDER BY score DESC
    LIMIT 20
       |
       v
  Doris Inverted Index -> raw hits
       |
       v
  BestMatchScorer -> re-rank with field boost + recency
       |
       v
  HighlightGenerator -> mark matched tokens
       |
       v
  FacetQueryBuilder -> parallel facet aggregations
       |
       v
  SearchResponse {
    hits: [{
      object_type: "Customer",
      object_id: "cust-001",
      score: 0.95,
      attributes: {display_name: "Zhang San", ...},
      highlights: [{field: "display_name",
                    fragment: "<em>Zhang San</em>"}]
    }, ...],
    total_count: 42,
    facets: [
      {field: "object_type", buckets: [
        {key: "Customer", count: 30},
        {key: "Order", count: 12}
      ]}
    ]
  }
       |
       v
  HotQueryManager.record("customer Zhang San", user_id)