purushah opened a new pull request, #1143: URL: https://github.com/apache/flink-agents/pull/1143
Closes #1102. Adds a Java vector store backed by PostgreSQL with the [pgvector](https://github.com/pgvector/pgvector) extension, so an existing PostgreSQL installation can serve as the vector store. Java already ships Elasticsearch, OpenSearch, Milvus and S3 Vectors; Python callers reach this store through the existing `JavaVectorStore` wrapper. ## What's in the change - New module `integrations/vector-stores/pgvector` with `PgVectorVectorStore` implementing `BaseVectorStore` and `CollectionManageableVectorStore`, following `MilvusVectorStore`. The only new third-party dependency is the PostgreSQL JDBC driver (`org.postgresql:postgresql:42.7.7`, property `postgresql.version` in `integrations/pom.xml`; the driver and the OnGres SCRAM/stringprep libraries it shades are listed in the dist `NOTICE` with their BSD 2-Clause license files). Pre-existing tables are usable as they are: a metadata cell that is not a JSON object is returned under `value`, and the ensure call inspects an existing table once per width/metric/index and warns on mismatches. - Arguments follow the existing stores: `uri` (JDBC URL) or `host`/`port`/`database`, `username`, `password`, `connect_timeout_s`, `socket_timeout_s`, `schema`, `collection` (table; `collection_name`/`index` aliases), `id_field`, `content_field`, `metadata_field`, `vector_field`, `dims`, `metric_type` (`COSINE`, `L2`, `IP`), `index_type` (`HNSW`, `IVFFLAT`, `NONE`), `index_params`, `create_extension`, `iterative_scan`. Credentials are not exposed through `getStoreKwargs()`; `user`/`password` query parameters embedded in `uri` are stripped there too. - Each collection is a table (`text` id primary key, `text` content, `jsonb` metadata, `vector(dims)` embedding) with a GIN index on the metadata column and the vector index for the metric's operator class. `createCollectionIfNotExists` runs as one transaction under advisory locks (one for `CREATE EXTENSION`, one per table) so parallel subtasks cannot race past `IF NOT EXISTS`; it rejects HNSW/IVFFlat above pgvector's 2000-dimension index limit only when it would create the table (existing wider tables can still be ensured and queried), and remembers tables it has created or verified (per requested width/metric/index, forgotten on `deleteCollection`) so repeated ensure calls cost nothing, and leaves a table it finds untouched (no index builds on someone else's table); after either path it checks the catalog and logs a warning when the vector width or index operator class does not match the store's `dims`/`metric_type`, or when `CREATE INDEX IF NOT EXISTS` was skipped because an un related relation already carried the index name. Equality filters compile to one `jsonb` containment predicate and apply to `query`, `get` and `delete`. Because PostgreSQL post-filters index candidates, on pgvector 0.8+ every similarity search runs as `BEGIN; SET LOCAL hnsw.iterative_scan ...; SET LOCAL ivfflat.iterative_scan ...; <search>; COMMIT` in a single `execute()` (`iterative_scan`, default `RELAXED_ORDER`, overridable per query), so it costs no extra round trip, leaves no session state and raises no server warning; a failed search rolls the block back so the cached connection stays usable so the index keeps scanning until `limit` rows match; older pgvector falls back to the plain scan. Hits are re-sorted through a materialized CTE, so results stay nearest-first even under relaxed ordering. The live test forces the HNSW post-filter plan through pgjdbc's `options` startup parameter and shows `off` coming up short while the default finds every match. `get` with a null limit re turns every matching row, per the `BaseVectorStore` contract. `add` and `update` both upsert on the primary key, as the other Java stores do, so a batch replayed after a Flink restart is idempotent; `add` generates ids that are missing, `update` requires them; a batch naming the same id twice stores the last document once (safe under pgjdbc's `reWriteBatchedInserts`); each batch is one transaction. A closed store refuses to reconnect. One JDBC connection per store instance is cached and reopened after connection-level failures; rows whose vector is `NULL` carry no score instead of a perfect one. - Score semantics per metric: cosine similarity for `COSINE` and inner product for `IP` (higher is better), Euclidean distance for `L2` (lower is better); results are nearest first. A per-query `metric_type` overrides the metric. - All identifiers (schema, table, columns, index parameter keys) are validated as `[A-Za-z_][A-Za-z0-9_]*` of at most 63 characters and quoted; index names are derived so that a long table name cannot make the metadata and vector index names collide after PostgreSQL's 63-character truncation (found by the live tests). - `dist` pins `org.checkerframework:checker-qual` to 3.49.3 (the newest requested: pgjdbc asks for 3.49.3, the Milvus SDK for 3.37.0) and updates `NOTICE` accordingly, since the bundled version otherwise depended on declaration order. - Registration: `ResourceName.VectorStore.PGVECTOR_VECTOR_STORE` (Java and the Python mirror), YAML alias `pgvector` in both alias tables with tests, `dist`, `ide-support` and the e2e integration module. - Live test infrastructure: `tools/docker/pgvector/docker-compose.yml` (`pgvector/pgvector:0.8.6-pg16`, pinned like the sibling compose files; host port overridable with `PGVECTOR_PORT`); `cross_language_tests` in CI starts the container (`docker compose up -d --wait`, on the compose healthcheck) for `PgVectorVectorStoreTest` and stops it right after, and `java_it_tests` starts it and exports `PGVECTOR_*` so the `PGVECTOR` e2e case runs, mirroring the Elasticsearch and Milvus steps. A Python cross-language e2e case for pgvector (like the Milvus one) is left as a follow-up. - E2E: `VectorStoreIntegrationTest` gains a `PGVECTOR` case gated on `PGVECTOR_URI` and on the Ollama model being available. The test seeds a random table before the job through the same `PgVectorVectorStore` descriptor the agent declares (create table, insert two rows with fixed vectors), the agent stays a pure retrieval agent whose Ollama embedding is exercised on the query path, and the same store drops the table afterwards. - Docs: a PostgreSQL pgvector section in `vector_stores.md` (parameters, table layout, filter and score semantics, usage), the collection-management and cross-language provider lists, the FAQ support matrix and the YAML alias table. Out of scope, as in the issue: keyword and hybrid query modes, `halfvec`, a Python-native implementation. Three review suggestions were deliberately not taken: relaxing the identifier rule to "anything, double-quoted" (names stay `[A-Za-z_][A-Za-z0-9_]*`, at most 63 characters, so they are also safe in advisory-lock keys and derived index names), a connection pool per store (each instance holds one connection and serialises its operations, as documented; async retrieval fan-out across a pool can follow once needed), and treating an empty id list in `get`/`delete` as "no id filter" like Milvus (here it returns nothing / deletes nothing, which is the safer reading of an ambiguous contract and is documented). ## Test commands and results Module tests (6 unit tests always; 9 live tests when `PGVECTOR_URI` is set), against `pgvector/pgvector:pg16` from the compose file: ``` PGVECTOR_PORT=55432 docker compose -f tools/docker/pgvector/docker-compose.yml up -d PGVECTOR_URI=jdbc:postgresql://localhost:55432/postgres PGVECTOR_USERNAME=postgres PGVECTOR_PASSWORD=postgres \ mvn -pl integrations/vector-stores/pgvector test # Tests run: 15, Failures: 0, Errors: 0, Skipped: 0 -- BUILD SUCCESS mvn -pl integrations/vector-stores/pgvector test # without the variable # Tests run: 15, Failures: 0, Errors: 0, Skipped: 9 -- BUILD SUCCESS ``` Alias tables: ``` mvn -pl api test -Dtest=AliasesTest # Tests run: 10, Failures: 0 cd python && uv run --no-sync pytest flink_agents/api # 276 passed, 12 skipped uv run --no-sync python ../e2e-test/test-scripts/check_resource_consistency.py # successful ``` Complete Java unit suite (all non-e2e modules, as `tools/ut.sh` runs it): ``` mvn test -pl '!e2e-test/flink-agents-end-to-end-tests-integration,!e2e-test/flink-agents-end-to-end-tests-resource-cross-language' # 19 modules: Tests run: 2364, Failures: 0, Errors: 0, Skipped: 51 -- BUILD SUCCESS (the 9 live pgvector tests are among the skipped without PGVECTOR_URI) ``` Complete Python unit suite: ``` cd python && uv run --no-sync pytest flink_agents -k "not e2e_tests" -m "not integration" # 1401 passed, 13 skipped, 172 deselected ``` End-to-end, embedded Flink 2.3 cluster, real pgvector container and local Ollama `nomic-embed-text` embeddings: ``` PGVECTOR_URI=jdbc:postgresql://localhost:55432/postgres mvn -pl e2e-test/flink-agents-end-to-end-tests-integration test -Dtest=VectorStoreIntegrationTest -Pflink-2.3 # [TEST] Vector store retrieval PASSED, count=2 # Tests run: 2, Failures: 0, Errors: 0, Skipped: 1 (ELASTICSEARCH skipped, no ES_HOST) -- BUILD SUCCESS ``` End-to-end, standalone local Flink 2.3.0 cluster (`start-cluster.sh`, `flink-agents-dist-flink-2.3` in `lib/`, throwaway job submitted with `flink run` that creates the table, inserts three documents with real embeddings, runs a semantic query, a filtered query and a lookup by id, and writes the results with a `FileSink`): ``` question=How does Flink process streams?|top=flink(0.777)|second=routing(0.553)|filtered=routing|byId=database question=What does pgvector add to PostgreSQL?|top=pgvector(0.796)|second=flink(0.543)|filtered=routing|byId=database # job FINISHED, no exceptions in JobManager/TaskManager logs, 3 rows in the table ``` Also run: `ruff check`, Spotless, Hugo build of the docs. Twenty-one local `/code-review` rounds were applied before pushing (`PGSimpleDataSource` carrying URL/credentials/timeouts, transactional batches, advisory-locked DDL, connection caching without a per-call validation round trip, credential stripping, NULL-vector scores, iterative index scans for filtered searches, unbounded null `get` limit, driver bumped past CVE-2025-49146, NOTICE/LICENSE entries). ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Claude Code 2.1.273 (Claude Fable 5.1) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
