Add support for bounded JDBC cursor

Author: exAspArk

server/client_error.go

@@ -11,6 +11,7 @@ const (
 	PG_ERROR_CODE_FEATURE_NOT_SUPPORTED               = "0A000"
 	PG_ERROR_CODE_PROTOCOL_VIOLATION                  = "08P01"
 	PG_ERROR_CODE_INVALID_PARAMETER_VALUE             = "22023"
+	PG_ERROR_CODE_INVALID_BINARY_REPRESENTATION       = "22P03"
 )
 
 type ClientError struct {

server/config.go

@@ -9,7 +9,7 @@ import (
 )
 
 const (
-	VERSION = "2.1.0"
+	VERSION = "2.2.0"
 
 	ENV_LOG_LEVEL     = "LOG_LEVEL"
 	ENV_AUDIT_ENABLED = "AUDIT_ENABLED"

server/duckdb_client.go

@@ -141,11 +141,6 @@ func (client *DuckdbClient) QueryContext(ctx context.Context, query string, args
 	return client.Db.QueryContext(ctx, query, args...)
 }
 
-func (client *DuckdbClient) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
-	LogDebug(client.Config, "Preparing DuckDB statement:", query)
-	return client.Db.PrepareContext(ctx, query)
-}
-
 func (client *DuckdbClient) ExecContext(ctx context.Context, query string, args ...map[string]string) (sql.Result, error) {
 	LogDebug(client.Config, "Executing DuckDB:", query)
 	if len(args) == 0 {
@@ -185,19 +180,6 @@ func (client *DuckdbClient) Close() {
 	client.Connector.Close()
 }
 
-func (client *DuckdbClient) RecreateDb() error {
-	db, connector, err := client.openDatabase()
-	if err != nil {
-		return fmt.Errorf("failed to recreate DuckDB: %w", err)
-	}
-
-	client.Db.Close()
-	client.Connector.Close()
-	client.Db = db
-	client.Connector = connector
-	return nil
-}
-
 func replaceNamedStringArgs(query string, args map[string]string) string {
 	for key, value := range args {
 		query = strings.ReplaceAll(

server/duckdb_client_test.go

@@ -51,10 +51,6 @@ func TestSearchReaderSeesExternalCommits(t *testing.T) {
 		t.Fatal(err)
 	}
 	check("after")
-	if err := reader.RecreateDb(); err != nil {
-		t.Fatal(err)
-	}
-	check("after")
 	if _, err := writer.ExecContext(ctx, "UPDATE search_store.main.products SET name = 'latest'"); err != nil {
 		t.Fatal(err)
 	}

server/embedding_client_test.go

@@ -65,22 +65,24 @@ func TestEmbeddingClientAndLiteralCache(t *testing.T) {
 	prepared := &PreparedStatement{VectorParameters: map[int]SearchParameterSpec{
 		1: {Dimensions: 2, RequireNonZero: true},
 	}}
-	_, prepared, err = queryHandler.HandleBindQuery(&pgproto3.Bind{Parameters: [][]byte{[]byte("dog")}}, prepared)
+	_, portal1, err := queryHandler.HandleBindQuery(context.Background(), &pgproto3.Bind{Parameters: [][]byte{[]byte("dog")}}, prepared)
+	defer portal1.Close()
 	if err != nil {
 		t.Fatal(err)
 	}
-	vector := prepared.Variables[0].([]float32)
+	vector := portal1.Variables[0].([]float32)
 	if len(vector) != 2 || vector[0] != 0.6 || vector[1] != 0.8 {
 		t.Fatalf("unexpected bound embedding: %#v", vector)
 	}
 	prepared = &PreparedStatement{VectorParameters: map[int]SearchParameterSpec{
 		1: {Dimensions: 2, RequireNonZero: true, LiteralText: &dog},
 	}}
-	_, prepared, err = queryHandler.HandleBindQuery(&pgproto3.Bind{}, prepared)
+	_, portal2, err := queryHandler.HandleBindQuery(context.Background(), &pgproto3.Bind{}, prepared)
+	defer portal2.Close()
 	if err != nil {
 		t.Fatal(err)
 	}
-	vector = prepared.Variables[0].([]float32)
+	vector = portal2.Variables[0].([]float32)
 	if len(vector) != 2 || vector[0] != 0.6 || vector[1] != 0.8 {
 		t.Fatalf("unexpected hidden literal embedding: %#v", vector)
 	}

server/main_test.go

@@ -54,7 +54,7 @@ func TestRuntimeManagerSharesLazyInitialization(t *testing.T) {
 	}
 }
 
-func TestRuntimeManagerWaitsForExpiredRuntimeToCloseBeforeRecreating(t *testing.T) {
+func TestRuntimeManagerReplacesExpiredRuntimeWhileOldReadersFinish(t *testing.T) {
 	var attempts atomic.Int32
 	runtime := fakeServerRuntime()
 	manager := &RuntimeManager{
@@ -101,14 +101,24 @@ func TestRuntimeManagerWaitsForExpiredRuntimeToCloseBeforeRecreating(t *testing.
 	}()
 
 	select {
-	case <-acquireDone:
+	case err := <-acquireDone:
+		if err != nil {
+			runtime.RequestMu.RUnlock()
+			t.Fatal(err)
+		}
+	case <-time.After(time.Second):
 		runtime.RequestMu.RUnlock()
-		t.Fatal("new runtime was created before expired runtime closed")
+		t.Fatal("new queries blocked behind retired runtime")
+	}
+	select {
+	case <-expired:
+		runtime.RequestMu.RUnlock()
+		t.Fatal("old runtime closed with active reader")
 	default:
 	}
-	if got := attempts.Load(); got != 0 {
+	if got := attempts.Load(); got != 1 {
 		runtime.RequestMu.RUnlock()
-		t.Fatalf("runtime creation attempts before close = %d, want 0", got)
+		t.Fatalf("initializations = %d", got)
 	}
 
 	runtime.RequestMu.RUnlock()
@@ -117,12 +127,16 @@ func TestRuntimeManagerWaitsForExpiredRuntimeToCloseBeforeRecreating(t *testing.
 	case <-time.After(time.Second):
 		t.Fatal("expired runtime did not close")
 	}
-	if err := <-acquireDone; err != nil {
-		t.Fatalf("runtime acquisition after close failed: %v", err)
-	}
 	if got := attempts.Load(); got != 1 {
 		t.Fatalf("runtime creation attempts after close = %d, want 1", got)
 	}
+	manager.mu.Lock()
+	replacement := manager.runtime
+	manager.mu.Unlock()
+	if replacement == nil || replacement == runtime {
+		t.Fatal("closing the retired runtime removed its replacement")
+	}
+
 }
 
 func TestRuntimeManagerRetriesAfterInitializationFailure(t *testing.T) {