support for PREPARE and EXECUTE statement

[TwistingTwists]

Sister PR: sunng87/pgwire#409

Closes: #292

-- Run inside psql connected to datafusion-postgres:
--   psql -h 127.0.0.1 -p 15432

-- 1. Basic PREPARE/EXECUTE
PREPARE my_stmt AS SELECT 42 AS answer;
EXECUTE my_stmt;
EXECUTE my_stmt;

-- 2. Parameterized
PREPARE param_stmt (INT) AS SELECT $1 * 2 AS doubled;
EXECUTE param_stmt(21);
EXECUTE param_stmt(100);

-- 3. DEALLOCATE specific (my_stmt gone, param_stmt still works)
DEALLOCATE my_stmt;
EXECUTE param_stmt(5);

-- 4. DEALLOCATE ALL
PREPARE stmt_a AS SELECT 1;
PREPARE stmt_b AS SELECT 2;
DEALLOCATE ALL;

-- 5. These should all error with "does not exist"
EXECUTE stmt_a;
EXECUTE stmt_b;
EXECUTE param_stmt(1);
EXECUTE never_prepared;

Prepared Statement Test Results

All tests passed end-to-end with real psql.

Test	Result
PREPARE my_stmt AS SELECT 42	✅ PREPARE
EXECUTE my_stmt (twice)	✅ Returns 42 both times
PREPARE param_stmt (INT) AS SELECT $1 * 2	✅ PREPARE
EXECUTE param_stmt(21)	✅ Returns 42
EXECUTE param_stmt(100)	✅ Returns 200
DEALLOCATE my_stmt	✅ DEALLOCATE
EXECUTE my_stmt after deallocate	✅ Error: "does not exist"
DEALLOCATE ALL	✅ DEALLOCATE
EXECUTE stmt_a after DEALLOCATE ALL	✅ Error: "does not exist"

[TwistingTwists]

I think first release of pgwire needs to be made and then update that here ?
This comes from testing locally.

relevant branch - sunng87/pgwire#409

[TwistingTwists]

sequenceDiagram
      participant U as User SQL Statement
      participant H as PrepareExecuteHook
      participant E as SessionExtensions
      participant O as Outer Lock (extensions map)
      participant P as Arc<PreparedStatements>
      participant I as Inner Lock (PreparedStatements)

      U->>H: PREPARE / EXECUTE / DEALLOCATE
      H->>E: statements()
      E->>O: acquire read lock
      alt Type entry exists
          O-->>E: found
          E-->>O: release read lock
      else Type entry missing
          O-->>E: not found
          E-->>O: release read lock
          E->>O: acquire write lock
          O-->>E: insert/get via or_insert_with
          E-->>O: release write lock
      end
      E-->>H: Arc<PreparedStatements>

      alt PREPARE / DEALLOCATE (mutating statements)
          H->>I: acquire write lock
          I-->>H: insert/remove/update
          H-->>I: release write lock
      else EXECUTE (reading statement)
          H->>I: acquire read lock
          I-->>H: lookup by statement name
          H-->>I: release read lock
      end

Loading

Putting this here -> may be this helps to visualise flow easier. It certainly helped me.

[TwistingTwists]

Hey @sunng87 ,
Would you be able to review this? I'll resolve conflicts in upcoming week.

[sunng87]

@TwistingTwists Sorry for delay. I'm still trying to figure out a better way to support this in upstream pgwire API.

[TwistingTwists]

Hey @sunng87 , now that upstream PR is merged, do you think this PR is roughly in the right direction?

[sunng87]

Hi @TwistingTwists , with latest pgwire, we will be able to use downcast to get portalstore work for simplequeryhandler. So I think we are now able to implement it just like we did with extended query.

[TwistingTwists]

Ah. Awesome 😎. Will update this PR with those changes then. :)