Method execute_dml

src/query/extended.rs:5072–5114

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

    }
    
    async fn execute_dml<T>(
        framed: &mut Framed<T, crate::protocol::PostgresCodec>,
        db: &Arc<DbHandler>,
        query: &str,
        portal_name: &str,
        session: &Arc<SessionState>,
    ) -> Result<(), PgSqliteError>
    where
        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
    {
        // Check for RETURNING clause
        if ReturningTranslator::has_returning_clause(query) {
            debug!("Extended protocol: Query has RETURNING clause, using execute_dml_with_returning: {}", query);
            // Get result formats from portal
            let result_formats = {
                let portals = session.portals.read().await;
                let portal = portals.get(portal_name).unwrap();
                portal.result_formats.clone()
            };
            return Self::execute_dml_with_returning(framed, db, session, query, &result_formats).await;
        }
        
        // Validation is now done in handle_execute before parameter substitution
        
        debug!("Extended protocol: Executing DML query without RETURNING: {}", query);
        let cached_conn = Self::get_or_cache_connection(session, db).await;
        let response = db.execute_with_session_cached(query, &session.id, cached_conn.as_ref()).await?;
        
        let tag = if query_starts_with_ignore_case(query, "INSERT") {
            format!("INSERT 0 {}", response.rows_affected)
        } else if query_starts_with_ignore_case(query, "UPDATE") {
            format!("UPDATE {}", response.rows_affected)
        } else if query_starts_with_ignore_case(query, "DELETE") {
            format!("DELETE {}", response.rows_affected)
        } else {
            format!("OK {}", response.rows_affected)
        };
        
        framed.send(BackendMessage::CommandComplete { tag }).await
            .map_err(PgSqliteError::Io)?;
        
        Ok(())
    }
    
    /// Helper function to build field descriptions for RETURNING clause with proper type detection
    async fn build_returning_field_descriptions(