[KM-520] Integrate db query executor pipeline with existing rag retrieve pipeline

src/api/v1/chat.py

@@ -9,6 +9,9 @@ from src.db.postgres.models import ChatMessage, MessageSource
 from src.agents.orchestration import orchestrator
 from src.agents.chatbot import chatbot
 from src.rag.retriever import retriever
+from src.rag.base import RetrievalResult
+from src.query.query_executor import query_executor
+from src.query.base import QueryResult
 from src.db.redis.connection import get_redis
 from src.config.settings import settings
 from src.middlewares.logging import get_logger, log_execution
@@ -88,6 +91,22 @@ def _extract_sources(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     return sources
 
 
+def _format_query_results(results: list[QueryResult]) -> str:
+    if not results:
+        return ""
+    lines = []
+    for r in results:
+        name = r.metadata.get("client_name", r.source_id)
+        lines.append(f"[Query result — {name}, tables: {r.table_or_file}]")
+        lines.append(f"SQL: {r.metadata.get('sql', '')}")
+        if r.columns and r.rows:
+            lines.append(" | ".join(r.columns))
+            for row in r.rows[:20]:
+                lines.append(" | ".join(str(row.get(c, "")) for c in r.columns))
+        lines.append(f"({r.row_count} rows total)\n")
+    return "\n".join(lines)
+
+
 async def get_cached_response(redis, cache_key: str) -> Optional[str]:
     cached = await redis.get(cache_key)
     if cached:
@@ -208,6 +227,27 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
             context = _format_context(raw_results)
             sources = _extract_sources(raw_results)
 
+            source_hint = intent_result.get("source_hint", "both")
+            if source_hint in ("schema", "both"):
+                retrieval_objects = [
+                    RetrievalResult(
+                        content=r["content"],
+                        metadata=r["metadata"],
+                        score=0.0,
+                        source_type=r["metadata"].get("source_type", ""),
+                    )
+                    for r in raw_results
+                ]
+                query_results = await query_executor.execute(
+                    results=retrieval_objects,
+                    user_id=request.user_id,
+                    db=db,
+                    question=request.message,
+                )
+                query_context = _format_query_results(query_results)
+                if query_context:
+                    context = query_context + "\n\n" + context
+
         # Step 3: Direct response for greetings / non-document intents
         if intent_result.get("direct_response"):
             response = intent_result["direct_response"]