haystack and pgbouncer

[MartinMayoIglesias]

There seem to be a problem with the interaction of PgvectorDocumentStore with pgbouncer. When one attempts to run the indexing pipeline to write on the database it raises an exception with error message "psycopg.errors.DuplicatePreparedStatement: prepared statement "_pg3_0" already exists"

[julian-risch]

Hello @MartinMayoIglesias could you provide more details on your setup please? What version of pgbouncer are you using? Could you share some code to reproduce the error? If you believe it is a bug in the implementation of the Haystack integration, please open an issue in the haystack-core-integrations repository: https://github.com/deepset-ai/haystack-core-integrations/issues/new?assignees=&labels=bug&projects=&template=bug_report.md&title=

[MartinMayoIglesias]

Hi, thanks for your reply.
Regarding your first question: we are running azure flexible server with pgbouncer, so we do not have too much wiggleroom with that config.
Regarding the second question, I upload here a toy example that should reproduce the error. It is after the second time this script is run that the above exception is raised (but important: you must re run it immediately, after a few seconds it is possible to re run it without any problem). When I remove the pgbouncer interaction by connecting to a different port (5432) directly to the DB everything works fine:

`from haystack import Pipeline
from haystack.components.converters import TextFileToDocument
from haystack.components.embedders import OpenAIDocumentEmbedder
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack.dataclasses import ByteStream
from haystack.document_stores.types import DuplicatePolicy
from haystack.utils import Secret
from haystack_integrations.components.retrievers.pgvector import PgvectorEmbeddingRetriever
from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore

api_key = "OPENAI_KEY"
connection_string="POSTGRES_CONNECTION_STRING_TO_PORT_6432"
table_name="TABLENAME"
document_embedder = OpenAIDocumentEmbedder(
model="text-embedding-3-small",
api_key=Secret.from_token(api_key)
)
document_store = PgvectorDocumentStore(
connection_string=Secret.from_token(connection_string),
table_name=table_name,
embedding_dimension=1536,
keyword_index_name="keyword_value"
)
embedding_retriever = PgvectorEmbeddingRetriever(document_store=document_store)
doc_writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.SKIP)
doc_splitter = DocumentSplitter()
indexing_pipeline = Pipeline()
indexing_pipeline.add_component("converter", TextFileToDocument())
indexing_pipeline.add_component("cleaner", DocumentCleaner())
indexing_pipeline.add_component("splitter", doc_splitter)
indexing_pipeline.add_component(instance=document_embedder, name="doc_embedder")
indexing_pipeline.add_component(instance=doc_writer, name="doc_writer")
indexing_pipeline.connect("converter", "cleaner")
indexing_pipeline.connect("cleaner", "splitter")
indexing_pipeline.connect("splitter", "doc_embedder.documents")
indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")

if name == "main":
text = "this is a test"
data = {
"converter": {
"sources": [ByteStream.from_string(text)],
}
}
indexing_pipeline.run(data=data)` # here the exception is raised