feat: cursor and fetch support (#339)

* feat: cursor and fetch support

* feat: update cursor to support extended query

* feat: use pgwire 0.40 fetch api

* test: update tests for empty fetch

* test: add an integration test for foreign data wrapper

* ci: install podman

* test: refactor integration test to use flake environment

* ci: remove podman info

* chore: add back podman info

Author: sunng87

.github/workflows/ci.yml

@@ -74,15 +74,12 @@ jobs:
     needs: [test]
     steps:
     - uses: actions/checkout@v6
-    - uses: actions-rs/toolchain@v1
-      with:
-        toolchain: stable
-        override: true
-    - run: |
-        pip install psycopg
+    - uses: cachix/install-nix-action@v31
+    - name: Initialize podman
+      run: nix develop --command podman info >/dev/null 2>&1 || true
     - run: |
         cd tests-integration
-        ./test.sh
+        nix develop --command bash test.sh
 
   msrv:
     name: MSRV

Cargo.lock

@@ -19,7 +19,7 @@ bytes = "1.11.1"
 chrono = { version = "0.4", features = ["std"] }
 datafusion = { version = "53" }
 futures = "0.3"
-pgwire = { version = "0.39", default-features = false }
+pgwire = { version = "0.40", default-features = false }
 postgres-types = "0.2"
 rust_decimal = { version = "1.42", features = ["db-postgres"] }
 tokio = { version = "1", default-features = false }

Cargo.toml

@@ -27,7 +27,7 @@ datafusion = { workspace = true, optional = true }
 futures.workspace = true
 geoarrow = { version = "0.8", optional = true }
 geoarrow-schema = { version = "0.8", optional = true }
-pg_interval = { version = "0.5.1", package = "pg_interval_2" }
+pg_interval = { version = "0.5.0" }
 pgwire = { workspace = true, default-features = false, features = ["server-api", "pg-ext-types"] }
 postgres-types.workspace = true
 rust_decimal.workspace = true

arrow-pg/Cargo.toml

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use datafusion::sql::sqlparser::ast::Statement;
 use datafusion::sql::sqlparser::dialect::PostgreSqlDialect;
+use datafusion::sql::sqlparser::keywords::Keyword;
 use datafusion::sql::sqlparser::parser::Parser;
 use datafusion::sql::sqlparser::parser::ParserError;
 use datafusion::sql::sqlparser::tokenizer::Token;
@@ -247,10 +248,19 @@ impl PostgresCompatibilityParser {
 
         // Get token values (without spans) and filter out only whitespace
         // Keep semicolons as they separate statements
+        // Also rewrite ABORT to ROLLBACK for postgres compatibility
+        // remove this when https://github.com/apache/datafusion-sqlparser-rs/pull/2332 is ready
         let filtered_tokens: Vec<Token> = tokens
             .iter()
             .map(|t| t.token.clone())
             .filter(|t| !matches!(t, Token::Whitespace(_)))
+            .map(|t| {
+                if matches!(&t, Token::Word(w) if w.keyword == Keyword::ABORT) {
+                    Token::make_keyword("ROLLBACK")
+                } else {
+                    t
+                }
+            })
             .collect();
 
         // Handle empty input