Method execute_select

src/query/executor.rs:963–1521

(
        framed: &mut Framed<T, crate::protocol::PostgresCodec>,
        db: &Arc<DbHandler>,
        session: &Arc<SessionState>,
        query: &str,
        translation_metadata: &crate::translato

    }
    
    async fn execute_select<T>(
        framed: &mut Framed<T, crate::protocol::PostgresCodec>,
        db: &Arc<DbHandler>,
        session: &Arc<SessionState>,
        query: &str,
        translation_metadata: &crate::translator::TranslationMetadata,
        query_router: Option<&Arc<QueryRouter>>,
    ) -> Result<(), PgSqliteError>
    where
        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
    {
        // debug!("execute_select (non-ultra-simple) called with query: {}", query);
        // SQLAlchemy manages transactions explicitly - don't start implicit transactions
        // debug!("=== EXECUTE_SELECT CALLED with query: {}", query);
        
        // Check wire protocol cache first for cacheable queries
        if crate::cache::is_cacheable_for_wire_protocol(query)
            && let Some(cached_response) = crate::cache::WIRE_PROTOCOL_CACHE.get(query) {
                debug!("Wire protocol cache hit for query: {}", query);
                
                // Send cached row description
                framed.send(BackendMessage::RowDescription(cached_response.row_description.clone())).await
                    .map_err(PgSqliteError::Io)?;
                
                // Send cached data rows (already encoded)
                for encoded_row in &cached_response.encoded_rows {
                    // Send pre-encoded data directly
                    framed.get_mut().write_all(encoded_row).await
                        .map_err(PgSqliteError::Io)?;
                }
                
                // Send command complete
                let tag = format!("SELECT {}", cached_response.row_count);
                framed.send(BackendMessage::CommandComplete { tag }).await
                    .map_err(PgSqliteError::Io)?;
                
                return Ok(());
            }
        
        // Check if this is a catalog query first
        println!("EXECUTOR: About to call catalog interceptor with query: '{}'", query);
        let response = if let Some(catalog_result) = crate::catalog::CatalogInterceptor::intercept_query(query, db.clone(), Some(session.clone())).await {
            info!("Query intercepted by catalog handler");
            println!("EXECUTOR: Got catalog result, about to unwrap");
            let unwrapped = catalog_result?;
            println!("EXECUTOR: Unwrapped catalog result, columns: {}, rows: {}", unwrapped.columns.len(), unwrapped.rows.len());
            unwrapped
        } else {
            // Route query through query router if available
            if let Some(router) = query_router {
                router.execute_query(query, session).await.map_err(|e| PgSqliteError::Protocol(e.to_string()))?
            } else {
                let cached_conn = Self::get_or_cache_connection(session, db).await;
                db.query_with_session_cached(query, &session.id, cached_conn.as_ref()).await?
            }
        };
        
        // Extract table name from query to look up schema