QRun-IO/qbit-quick-search
3
0 Fork 0
Code Issues 2 Pull Requests 4 Actions Packages Projects Releases 14 Wiki Activity

feat: re-pin to qbit-build-parent 2.0.0, fix index drift, add reconcile (QRun-IO/qbit-quick-search#6) #7

Merged
KofTwentyTwo merged 9 commits from feature/GH-6-repin-and-drift-fix into develop 2026-09-26 01:36:57 +00:00
Conversation 4 Commits 9 Files Changed 29
+1693 -2401
29 changed files with 1693 additions and 2401 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.circleci/config.yml
+20
View File
@@ -3,10 +3,22 @@ version: 2.1
orbs:
qqq-orb: kingsrook/qqq-orb@0.5.2
commands:
configure-package-install:
steps:
- run:
name: Keep package installation unattended
command: |
# These build machines are discarded after the job.
sudo install -d /etc/needrestart/conf.d
printf '%s\n' '$nrconf{restart} = "l";' | sudo tee /etc/needrestart/conf.d/99-qqq-ci.conf >/dev/null
workflows:
test_only:
jobs:
- qqq-orb/mvn_test_only:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
filters:
branches:
@@ -15,6 +27,8 @@ workflows:
publish_snapshot:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: snapshot
filters:
@@ -24,6 +38,8 @@ workflows:
publish_release_candidate:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: release_candidate
filters:
@@ -33,6 +49,8 @@ workflows:
publish_release:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: release
filters:
@@ -44,6 +62,8 @@ workflows:
publish_hotfix_release:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: hotfix
filters:
.claude/settings.local.json
-63
View File
@@ -1,63 +0,0 @@
{
"permissions": {
"allow": [
"Bash(find:*)",
"Bash(grep:*)",
"Bash(mvn compile:*)",
"Bash(jar -tf:*)",
"Bash(jar -xf:*)",
"Bash(cat:*)",
"Bash(mvn test:*)",
"Bash(done)",
"Bash(mvn verify:*)",
"Bash(git add:*)",
"Bash(git commit:*)",
"Bash(git push:*)",
"mcp__circleci-mcp-server__get_latest_pipeline_status",
"mcp__circleci-mcp-server__list_followed_projects",
"mcp__circleci-mcp-server__run_pipeline",
"mcp__circleci-mcp-server__get_build_failure_logs",
"mcp__circleci-mcp-server__config_helper",
"mcp__github__get_file_contents",
"Bash(mvn dependency:tree:*)",
"Bash(mvn help:effective-pom:*)",
"Bash(git checkout:*)",
"Bash(ls /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/docs/*.md)",
"Bash(git:*)",
"Bash(ls:*)",
"Bash(open:*)",
"Bash(WT=/Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/.claude/worktrees/agent-a9dc6e8e)",
"Bash(DEST=/Users/james.maes/Git.Local/QRun-IO/qbit-quick-search)",
"Bash(cp \"$WT/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/BulkIndexResult.java\" \"$DEST/src/main/java/com/kingsrook/qbits/quicksearch/opensearch/\")",
"Bash(mvn clean:*)",
"Bash(jar tf:*)",
"Bash(unzip -p /Users/james.maes/.m2/repository/com/kingsrook/qqq/qqq-backend-core/0.40.0-SNAPSHOT/qqq-backend-core-0.40.0-SNAPSHOT.jar com/kingsrook/qqq/backend/core/actions/customizers/TableCustomizerInterface.java)",
"Bash(sed -i '' 's/postInsertRecord/postInsert/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java)",
"Bash(sed -i '' 's/postUpdateRecord/postUpdate/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java)",
"Bash(sed -i '' 's/postDeleteRecord/postDelete/g' src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java)",
"Bash(docker info:*)",
"Bash(docker run:*)",
"Read(//var/run/**)",
"Read(//Users/james.maes/**)",
"Bash(echo \"DOCKER_HOST=$DOCKER_HOST\")",
"Bash(echo \"TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=$TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE\")",
"Bash(DOCKER_HOST=unix:///var/run/docker.sock TESTCONTAINERS_RYUK_DISABLED=true mvn verify -Dit.test=OpenSearchIntegrationTest)",
"Bash(docker version:*)",
"Bash(DOCKER_HOST=unix:///var/run/docker.sock mvn verify -Dit.test=OpenSearchIntegrationTest)",
"Bash(curl -s --unix-socket /var/run/docker.sock http://localhost/v1.24/info)",
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} OS: {d.get\\(OperatingSystem,?\\)}''''\\)\")",
"Bash(curl -s --unix-socket /Users/james.maes/Library/Containers/com.docker.docker/Data/docker-cli.sock http://localhost/v1.47/info)",
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} OS: {d.get\\(OperatingSystem,?\\)} Containers: {d.get\\(Containers,?\\)}''''\\)\")",
"Bash(curl -s --unix-socket /Users/james.maes/.docker/run/docker.sock http://localhost/v1.47/info)",
"Bash(python3 -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\(f''''ServerVersion: {d.get\\(ServerVersion,?\\)} NCPU: {d.get\\(NCPU,?\\)} OSType: {d.get\\(OSType,?\\)}''''\\)\")",
"Bash(DOCKER_HOST=unix:///Users/james.maes/.docker/run/docker.sock DOCKER_API_VERSION=1.47 mvn verify -Dit.test=OpenSearchIntegrationTest -pl .)",
"Bash(cp /Users/james.maes/Git.Local/QRun-IO/qbit-webhooks/LICENSE /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/LICENSE)",
"Bash(cp -r /Users/james.maes/Git.Local/QRun-IO/qbit-webhooks/checkstyle/* /Users/james.maes/Git.Local/QRun-IO/qbit-quick-search/checkstyle/)",
"Bash(gh pr:*)",
"Bash(gh repo:*)",
"Bash(gh api:*)",
"Bash(python3 -c \":*)",
"mcp__circleci-mcp-server__get_job_test_results"
]
}
}
.gitignore
+1
View File
@@ -18,3 +18,4 @@ hs_err_pid*
dependency-reduced-pom.xml
/.env.local
/.cache/
.claude/settings.local.json
CLAUDE.md
+11 -9
View File
@@ -4,20 +4,22 @@ This file provides guidance to Claude Code when working with code in this reposi
## Project Overview
**QBit Quick Search** - A QQQ QBit providing OpenSearch-backed full-text search across QQQ applications. Current version: 0.1.0.
**QBit Quick Search** - A QQQ QBit providing OpenSearch-backed full-text search across QQQ applications.
## Build Commands
```bash
mvn compile # Compile
mvn test # Run all tests (174 tests)
mvn test # Run unit tests
mvn test -Dtest=ClassName # Run single test class
mvn package # Build JAR
mvn verify # Full verify with coverage check
mvn verify # Full verify, including integration tests
mvn verify -Pqqq-snapshot # Verify against 4.1.0-SNAPSHOT (override with -Dqqq.snapshot.version)
```
Integration tests (require Docker) run via `mvn verify` using maven-failsafe-plugin.
Unit tests exclude `*IntegrationTest.java` and `*IT.java`.
Integration tests use `@ExtendWith(RequiresDockerCondition.class)`: skipped without Docker locally, failed when `CI=true`.
## Architecture
@@ -32,13 +34,14 @@ Unit tests exclude `*IntegrationTest.java` and `*IT.java`.
**Customizers** (`customizers/` package, all three registered when `enableRealTimeIndexing = true`):
- `QuickSearchPostInsertCustomizer` - Fires after insert, publishes index events
- `QuickSearchPostUpdateCustomizer` - Fires after update, publishes index events
- `QuickSearchPostUpdateCustomizer` - Fires after update, re-reads the full records from the source table, publishes index events
- `QuickSearchPostDeleteCustomizer` - Fires after delete, publishes delete events
**Processes** (`processes/` package):
- `AbstractIndexingStep` - Base class with shared indexing logic (query source table, build documents, publish events, update run record)
- `BasepullIndexStep` - Queries records modified since last run timestamp
- `FullReindexStep` - Wipes and rebuilds the index for one or all tables
- `ReconcileIndexStep` - Re-indexes each table in primary-key order, then removes documents with no source record (no wipe, so no search blackout)
- `IndexingUtils` - Converts `QRecord` values to indexed text using field configs
**Publisher** (`publisher/` package):
@@ -68,12 +71,11 @@ Unit tests exclude `*IntegrationTest.java` and `*IT.java`.
## Dependencies
- Java 21
- QQQ Backend Core 0.40.0-SNAPSHOT (via `qqq-bom-pom`)
- Parent `com.kingsrook:qbit-build-parent:2.0.0`, the only source of the qqq version (QQQ 4.0.0); do not re-import `qqq-bom-pom` here (ADR-0007), except in the opt-in `qqq-snapshot` profile (`-Pqqq-snapshot`), which imports `qqq-bom-pom:${qqq.snapshot.version}` (default 4.1.0-SNAPSHOT) and adds the Central snapshots repository
- OpenSearch Java Client 2.10.0
- Apache HttpClient5 5.3
- jackson-datatype-jsr310 2.21.0 (Instant serialization for OpenSearch client)
- Apache HttpClient5 and jackson-datatype-jsr310, versions managed by the qqq BOM (keep them unpinned so they match its httpcore5 and jackson)
- JUnit 5 + AssertJ + Mockito for testing
- Testcontainers for integration tests (requires Docker)
- Testcontainers 1.21.4 for integration tests (requires Docker)
## Knowledge base
@@ -83,4 +85,4 @@ Second-brain vault notes covering this repo and the QQQ platform it plugs into:
- This repo's dossier: `$SECOND_BRAIN_VAULT/knowledge/qqq/repos/qbit-quick-search.md` (reviewed at develop @ `8d9ec2711fde`, 2026-07-04)
- Production-readiness audit + P0-P3 roadmap: `$SECOND_BRAIN_VAULT/projects/qbit-quick-search.md`
Note: some facts above in this file are stale (version is 0.2.1-SNAPSHOT, 215 tests, qqq dep is release 0.40.0 not SNAPSHOT) — trust the dossier.
Note: the dossier was reviewed before the qqq 4.0 re-pin; trust this file for build and dependency facts.
README.md
+32 -4
View File
@@ -35,7 +35,7 @@ QBit Quick Search is 100% open source under Apache 2.0. All data stays in your O
### Technology Stack
- **Java 21** with QQQ backend modules
- **Java 21** with QQQ 4.x backend modules (versions from `qbit-build-parent` 2.0.0)
- **OpenSearch 2.x** for full-text search with custom edge-ngram analyzer
- **QQQ Framework**: Entities, processes, customizers, permissions, API layer
@@ -54,7 +54,7 @@ qbit-quick-search/
model/ -- QuickSearchIndex, QuickSearchIndexRun entities
opensearch/ -- OpenSearch client, document model, bulk results
actions/ -- QuickSearchAction, input/output DTOs
processes/ -- AbstractIndexingStep, BasepullIndexStep, FullReindexStep
processes/ -- AbstractIndexingStep, BasepullIndexStep, FullReindexStep, ReconcileIndexStep
customizers/ -- Post-insert, post-update, post-delete customizers
publisher/ -- IndexEventPublisher interface, SynchronousIndexEventPublisher
```
@@ -253,6 +253,24 @@ new FullReindexStep().run(input, output);
Or through the QQQ admin UI via the "Full Reindex" process.
### Reconciling the Index
If the index may have drifted from the source tables (for example, deletes that
happened while OpenSearch was down), run the reconcile process. It re-indexes
every source record, then removes documents whose source record no longer
exists. It does not wipe the index first, so search keeps working while it runs.
```java
RunBackendStepInput input = new RunBackendStepInput();
input.addValue("tableName", "customer"); // optional: omit to reconcile all tables
RunBackendStepOutput output = new RunBackendStepOutput();
new ReconcileIndexStep().run(input, output);
// output values: recordsIndexed, documentsRemoved
```
Or through the QQQ admin UI via the "Reconcile Index" process. If any document
fails to index, the run is marked `FAILED` and no documents are removed.
## Data Model
### Tables (2)
@@ -262,12 +280,13 @@ Or through the QQQ admin UI via the "Full Reindex" process.
| `quickSearchIndex` | Per-table index configuration, status, and basepull tracking |
| `quickSearchIndexRun` | Individual indexing run history with record counts and error tracking |
### Processes (2)
### Processes (3)
| Process | Description |
|---------|-------------|
| Basepull Index | Scheduled: queries each source table for records modified since last run, indexes in batches |
| Full Reindex | On-demand: deletes and re-indexes all records for one or all tables |
| Reconcile Index | On-demand: re-indexes all records for one or all tables, then removes documents with no source record, without wiping the index first |
## Configuration Reference
@@ -324,11 +343,20 @@ Index `wmsItem` and `wmsLocation` for warehouse item and location search.
## Testing
```bash
mvn test # Run all 174 unit tests
mvn test # Run unit tests
mvn verify # Run unit tests + integration tests (requires Docker)
mvn test -Dtest=ClassName # Run single test class
mvn verify -Pqqq-snapshot # Verify against the next qqq line (4.1.0-SNAPSHOT)
```
The qqq version comes from `qbit-build-parent`. The opt-in `qqq-snapshot`
profile imports `qqq-bom-pom` at `${qqq.snapshot.version}` (default
`4.1.0-SNAPSHOT`) ahead of the parent's BOM and adds the Central snapshots
repository. Pick another version with `-Dqqq.snapshot.version=...`.
Without Docker, integration tests are skipped locally. With `CI=true`, they fail
instead, so a CI build cannot pass with its integration tests silently skipped.
### Coverage
- **70%+ instruction coverage** (JaCoCo enforced)
com/kingsrook/qqq/backend/core/actions/customizers/TableCustomizerInterface.java
-288
View File
@@ -1,288 +0,0 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2024. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.kingsrook.qqq.backend.core.actions.customizers;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.QueryOrGetInputInterface;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
/*******************************************************************************
** Common interface used by all (core) TableCustomizer types (e.g., post-query,
** and {pre,post}-{insert,update,delete}.
**
** Note that the abstract-base classes for each action still exist, though have
** been back-ported to be implementors of this interface. The action classes
** will now expect this type, and call this type's methods.
**
*******************************************************************************/
public interface TableCustomizerInterface
{
QLogger LOG = QLogger.getLogger(TableCustomizerInterface.class);
/*******************************************************************************
** custom actions to run after a query (or get!) takes place.
**
*******************************************************************************/
default List<QRecord> postQuery(QueryOrGetInputInterface queryInput, List<QRecord> records) throws QException
{
LOG.info("A default implementation of postQuery is running... Probably not expected!", logPair("tableName", queryInput.getTableName()));
return (records);
}
/*******************************************************************************
** custom actions before an insert takes place.
**
** It's important for implementations to be aware of the isPreview field, which
** is set to true when the code is running to give users advice, e.g., on a review
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
** things like storing data, you don't want to do that if isPreview is true!!
**
** General implementation would be, to iterate over the records (the inputs to
** the insert action), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
** - possibly manipulating values (`setValue`)
** - possibly throwing an exception - if you really don't want the insert operation to continue.
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) that you want to go on to the backend implementation class.
*******************************************************************************/
default List<QRecord> preInsert(InsertInput insertInput, List<QRecord> records, boolean isPreview) throws QException
{
try
{
return (preInsertOrUpdate(insertInput, records, isPreview, Optional.empty()));
}
catch(NotImplementedHereException e)
{
LOG.info("A default implementation of preInsert is running... Probably not expected!", logPair("tableName", insertInput.getTableName()));
return (records);
}
}
/*******************************************************************************
**
*******************************************************************************/
default AbstractPreInsertCustomizer.WhenToRun whenToRunPreInsert(InsertInput insertInput, boolean isPreview)
{
return (AbstractPreInsertCustomizer.WhenToRun.AFTER_ALL_VALIDATIONS);
}
/*******************************************************************************
** custom actions after an insert takes place.
**
** General implementation would be, to iterate over the records (the outputs of
** the insert action), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
** - possibly throwing an exception - though doing so won't stop the update, and instead
** will just set a warning on all of the updated records...
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) that you want to go back to the caller.
*******************************************************************************/
default List<QRecord> postInsert(InsertInput insertInput, List<QRecord> records) throws QException
{
try
{
return (postInsertOrUpdate(insertInput, records, Optional.empty()));
}
catch(NotImplementedHereException e)
{
LOG.info("A default implementation of postInsert is running... Probably not expected!", logPair("tableName", insertInput.getTableName()));
return (records);
}
}
/*******************************************************************************
** custom actions before an update takes place.
**
** It's important for implementations to be aware of the isPreview field, which
** is set to true when the code is running to give users advice, e.g., on a review
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
** things like storing data, you don't want to do that if isPreview is true!!
**
** General implementation would be, to iterate over the records (the inputs to
** the update action), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
** - possibly manipulating values (`setValue`)
** - possibly throwing an exception - if you really don't want the update operation to continue.
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) that you want to go on to the backend implementation class.
**
** Note, "old records" (e.g., with values freshly fetched from the backend) will be
** available (if the backend supports it)
*******************************************************************************/
default List<QRecord> preUpdate(UpdateInput updateInput, List<QRecord> records, boolean isPreview, Optional<List<QRecord>> oldRecordList) throws QException
{
try
{
return (preInsertOrUpdate(updateInput, records, isPreview, oldRecordList));
}
catch(NotImplementedHereException e)
{
LOG.info("A default implementation of preUpdate is running... Probably not expected!", logPair("tableName", updateInput.getTableName()));
return (records);
}
}
/*******************************************************************************
** custom actions after an update takes place.
**
** General implementation would be, to iterate over the records (the outputs of
** the update action), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records?
** - possibly throwing an exception - though doing so won't stop the update, and instead
** will just set a warning on all of the updated records...
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) that you want to go back to the caller.
**
** Note, "old records" (e.g., with values freshly fetched from the backend) will be
** available (if the backend supports it).
*******************************************************************************/
default List<QRecord> postUpdate(UpdateInput updateInput, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
{
try
{
return (postInsertOrUpdate(updateInput, records, oldRecordList));
}
catch(NotImplementedHereException e)
{
LOG.info("A default implementation of postUpdate is running... Probably not expected!", logPair("tableName", updateInput.getTableName()));
return (records);
}
}
/*******************************************************************************
** Custom actions before a delete takes place.
**
** It's important for implementations to be aware of the isPreview param, which
** is set to true when the code is running to give users advice, e.g., on a review
** screen - vs. being false when the action is ACTUALLY happening. So, if you're doing
** things like storing data, you don't want to do that if isPreview is true!!
**
** General implementation would be, to iterate over the records (which the DeleteAction
** would look up based on the inputs to the delete action), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records
** - possibly throwing an exception - if you really don't want the delete operation to continue.
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) - this is how errors
** and warnings are propagated to the DeleteAction. Note that any records with
** an error will NOT proceed to the backend's delete interface - but those with
** warnings will.
*******************************************************************************/
default List<QRecord> preDelete(DeleteInput deleteInput, List<QRecord> records, boolean isPreview) throws QException
{
LOG.info("A default implementation of preDelete is running... Probably not expected!", logPair("tableName", deleteInput.getTableName()));
return (records);
}
/*******************************************************************************
** Custom actions after a delete takes place.
**
** General implementation would be, to iterate over the records (ones which didn't
** have a delete error), and look at their values:
** - possibly adding Errors (`addError`) or Warnings (`addWarning`) to the records?
** - possibly throwing an exception - though doing so won't stop the delete, and instead
** will just set a warning on all of the deleted records...
** - doing "whatever else" you may want to do.
** - returning the list of records (can be the input list) that you want to go back
** to the caller - this is how errors and warnings are propagated .
*******************************************************************************/
default List<QRecord> postDelete(DeleteInput deleteInput, List<QRecord> records) throws QException
{
LOG.info("A default implementation of postDelete is running... Probably not expected!", logPair("tableName", deleteInput.getTableName()));
return (records);
}
/***************************************************************************
** Optional method to override in a customizer that does the same thing for
** both preInsert & preUpdate.
***************************************************************************/
default List<QRecord> preInsertOrUpdate(AbstractActionInput input, List<QRecord> records, boolean isPreview, Optional<List<QRecord>> oldRecordList) throws QException
{
throw NotImplementedHereException.instance;
}
/***************************************************************************
** Optional method to override in a customizer that does the same thing for
** both postInsert & postUpdate.
***************************************************************************/
default List<QRecord> postInsertOrUpdate(AbstractActionInput input, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
{
throw NotImplementedHereException.instance;
}
/***************************************************************************
**
***************************************************************************/
default Optional<Map<Serializable, QRecord>> oldRecordListToMap(String primaryKeyField, Optional<List<QRecord>> oldRecordList)
{
if(oldRecordList.isPresent())
{
return (Optional.of(CollectionUtils.listToMap(oldRecordList.get(), r -> r.getValue(primaryKeyField))));
}
else
{
return (Optional.empty());
}
}
/***************************************************************************
**
***************************************************************************/
class NotImplementedHereException extends QException
{
private static NotImplementedHereException instance = new NotImplementedHereException();
/***************************************************************************
**
***************************************************************************/
private NotImplementedHereException()
{
super("Not implemented here");
}
}
}
com/kingsrook/qqq/backend/core/model/metadata/SupplementalCustomizerType.java
-35
View File
@@ -1,35 +0,0 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2022. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.kingsrook.qqq.backend.core.model.metadata;
/*******************************************************************************
** Interface to be implemented by enums of the possible supplemental types
**
*******************************************************************************/
public interface SupplementalCustomizerType
{
/*******************************************************************************
** allows implementing enums to return the expected class being used
*******************************************************************************/
public Class<?> getCodeReferenceType();
}
com/kingsrook/qqq/backend/core/model/metadata/tables/QTableMetaData.java
-1908
View File
File diff suppressed because it is too large. Load diff
pom.xml
+40 -12
View File
@@ -27,7 +27,7 @@
<parent>
<groupId>com.kingsrook</groupId>
<artifactId>qbit-build-parent</artifactId>
<version>1.6.0</version>
<version>2.0.0</version>
</parent>
<groupId>com.kingsrook.qbits</groupId>
@@ -75,13 +75,6 @@
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>com.kingsrook.qqq</groupId>
<artifactId>qqq-bom-pom</artifactId>
<version>0.40.0</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
@@ -97,16 +90,16 @@
<version>${opensearch.version}</version>
</dependency>
<!-- httpclient5 and jackson versions are managed by the qqq BOM, so they -->
<!-- stay binary-compatible with the httpcore5 and jackson it brings. -->
<dependency>
<groupId>org.apache.httpcomponents.client5</groupId>
<artifactId>httpclient5</artifactId>
<version>5.3</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>2.21.0</version>
</dependency>
<dependency>
@@ -147,13 +140,13 @@
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>1.21.0</version>
<version>1.21.4</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>1.21.0</version>
<version>1.21.4</version>
<scope>test</scope>
</dependency>
</dependencies>
@@ -224,4 +217,39 @@
</plugin>
</plugins>
</build>
<profiles>
<!-- Opt-in check against the next qqq line: mvn -B verify -Pqqq-snapshot -->
<!-- (override the version with -Dqqq.snapshot.version=...). Default builds -->
<!-- take the qqq version only from qbit-build-parent (ADR-0007). -->
<profile>
<id>qqq-snapshot</id>
<properties>
<qqq.snapshot.version>4.1.0-SNAPSHOT</qqq.snapshot.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.kingsrook.qqq</groupId>
<artifactId>qqq-bom-pom</artifactId>
<version>${qqq.snapshot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<repositories>
<repository>
<id>sonatype-snapshots</id>
<url>https://central.sonatype.com/repository/maven-snapshots/</url>
<releases>
<enabled>false</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>
</profile>
</profiles>
</project>
src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducer.java
+23
View File
@@ -43,6 +43,7 @@ import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import com.kingsrook.qbits.quicksearch.processes.BasepullIndexStep;
import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher;
@@ -62,6 +63,7 @@ public class QuickSearchQBitProducer
public static final String BASEPULL_PROCESS_NAME = "quickSearchBasepullIndex";
public static final String FULL_REINDEX_PROCESS_NAME = "quickSearchFullReindex";
public static final String RECONCILE_PROCESS_NAME = "quickSearchReconcileIndex";
public static final String APP_NAME = "quickSearchAdmin";
private QuickSearchQBitConfig config;
@@ -217,14 +219,17 @@ public class QuickSearchQBitProducer
////////////////////////////////////////////////////
String basepullProcessName = config.applyPrefix(BASEPULL_PROCESS_NAME);
String fullReindexProcessName = config.applyPrefix(FULL_REINDEX_PROCESS_NAME);
String reconcileProcessName = config.applyPrefix(RECONCILE_PROCESS_NAME);
QProcessMetaData basepullProcess = buildBasepullProcess(basepullProcessName);
QProcessMetaData fullReindexProcess = buildFullReindexProcess(fullReindexProcessName);
QProcessMetaData reconcileProcess = buildReconcileProcess(reconcileProcessName);
if(Boolean.TRUE.equals(config.getEnableScheduledProcesses()))
{
qInstance.addProcess(basepullProcess);
qInstance.addProcess(fullReindexProcess);
qInstance.addProcess(reconcileProcess);
}
////////////////////////////////////////////////////
@@ -249,6 +254,7 @@ public class QuickSearchQBitProducer
{
app.withChild(basepullProcess);
app.withChild(fullReindexProcess);
app.withChild(reconcileProcess);
}
qInstance.addApp(app);
@@ -527,6 +533,23 @@ public class QuickSearchQBitProducer
/***************************************************************************
** Build the reconcile process metadata: re-index each table from its
** source and remove documents with no source record, without wiping the
** index first. Takes the same optional tableName input as full reindex.
***************************************************************************/
private QProcessMetaData buildReconcileProcess(String processName)
{
return new QProcessMetaData()
.withName(processName)
.withLabel("Quick Search Reconcile Index")
.withStep(new QBackendStepMetaData()
.withName("reconcile")
.withCode(new QCodeReference(ReconcileIndexStep.class)));
}
/***************************************************************************
** Register post-insert, post-update, and post-delete customizers on each
** discovered source table.
src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizer.java
+18 -4
View File
@@ -24,6 +24,7 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -56,8 +57,8 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
/*******************************************************************************
** Post-delete hook: builds DELETE events for each deleted record and publishes
** them via the configured IndexEventPublisher.
** Post-delete hook: builds DELETE events for each successfully deleted
** record and publishes them via the configured IndexEventPublisher.
**
** If the publisher is null (QBit not yet initialized) or if publishing fails,
** a warning is logged and the original records are returned without throwing,
@@ -87,13 +88,26 @@ public class QuickSearchPostDeleteCustomizer implements TableCustomizerInterface
List<IndexEvent> events = new ArrayList<>();
for(QRecord record : records)
{
///////////////////////////////////////////////////////////////////
// a record with errors was not deleted, so its document stays; //
// one without a primary key has no document id to remove //
///////////////////////////////////////////////////////////////////
String recordId = record.getValueString(primaryKeyField);
if(recordId == null || CollectionUtils.nullSafeHasContents(record.getErrors()))
{
continue;
}
events.add(new IndexEvent()
.withAction(IndexEventAction.DELETE)
.withTableName(tableName)
.withRecordId(record.getValueString(primaryKeyField)));
.withRecordId(recordId));
}
publisher.publishDeleteEvents(events);
if(!events.isEmpty())
{
publisher.publishDeleteEvents(events);
}
}
catch(Exception e)
{
src/main/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizer.java
+53 -4
View File
@@ -17,14 +17,21 @@
package com.kingsrook.qbits.quicksearch.customizers;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -37,6 +44,11 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
** Post-update table customizer that re-indexes updated records in the Quick
** Search OpenSearch index.
**
** An update's records may hold only the fields that changed, so each updated
** record is re-read from the source table (in the update's transaction) and
** the full record is indexed. Indexing the sparse record would drop the
** unchanged fields from the document.
**
** QQQ instantiates this class via reflection using the no-arg constructor.
** The publisher and table configuration are obtained from the static
** QuickSearchQBitContext at call time.
@@ -57,8 +69,10 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
/*******************************************************************************
** Post-update hook: builds INDEX events for each updated record and publishes
** them via the configured IndexEventPublisher.
** Post-update hook: re-reads the successfully updated records from the
** source table, builds INDEX events from them, and publishes them via the
** configured IndexEventPublisher. Records with errors were not updated, so
** they are skipped.
**
** If the publisher is null (QBit not yet initialized) or if publishing fails,
** a warning is logged and the original records are returned without throwing,
@@ -86,7 +100,7 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
String primaryKeyField = (tableConfig != null && tableConfig.getPrimaryKeyField() != null) ? tableConfig.getPrimaryKeyField() : "id";
List<IndexEvent> events = new ArrayList<>();
for(QRecord record : records)
for(QRecord record : fetchCurrentRecords(updateInput, records, primaryKeyField))
{
events.add(new IndexEvent()
.withAction(IndexEventAction.INDEX)
@@ -95,7 +109,10 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
.withRecord(record));
}
publisher.publishIndexEvents(events);
if(!events.isEmpty())
{
publisher.publishIndexEvents(events);
}
}
catch(Exception e)
{
@@ -105,4 +122,36 @@ public class QuickSearchPostUpdateCustomizer implements TableCustomizerInterface
return records;
}
/*******************************************************************************
** Query the source table for the full, current version of each successfully
** updated record. Uses the update's transaction, so uncommitted changes are
** visible.
*******************************************************************************/
private List<QRecord> fetchCurrentRecords(UpdateInput updateInput, List<QRecord> updatedRecords, String primaryKeyField) throws QException
{
List<Serializable> primaryKeys = new ArrayList<>();
for(QRecord record : updatedRecords)
{
Serializable primaryKey = record.getValue(primaryKeyField);
if(primaryKey != null && CollectionUtils.nullSafeIsEmpty(record.getErrors()))
{
primaryKeys.add(primaryKey);
}
}
if(primaryKeys.isEmpty())
{
return (new ArrayList<>());
}
QueryInput queryInput = new QueryInput();
queryInput.setTableName(updateInput.getTableName());
queryInput.setTransaction(updateInput.getTransaction());
queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, primaryKeys)));
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
}
}
src/main/java/com/kingsrook/qbits/quicksearch/opensearch/OpenSearchDocument.java
+10
View File
@@ -172,6 +172,16 @@ public class OpenSearchDocument
*******************************************************************************/
@JsonIgnore
public String getDocumentId()
{
return (buildDocumentId(sourceTable, recordId));
}
/*******************************************************************************
** Build the composite OpenSearch _id for a source table and record ID.
*******************************************************************************/
public static String buildDocumentId(String sourceTable, String recordId)
{
return (sourceTable + ":" + recordId);
}
src/main/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClient.java
+148 -27
View File
@@ -19,6 +19,7 @@ package com.kingsrook.qbits.quicksearch.opensearch;
import java.io.Closeable;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -32,8 +33,10 @@ import org.apache.hc.client5.http.auth.AuthScope;
import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
import org.apache.hc.core5.http.HttpHost;
import org.opensearch.client.json.JsonData;
import org.opensearch.client.json.jackson.JacksonJsonpMapper;
import org.opensearch.client.opensearch.OpenSearchClient;
import org.opensearch.client.opensearch._types.Conflicts;
import org.opensearch.client.opensearch._types.FieldValue;
import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
import org.opensearch.client.opensearch._types.query_dsl.MultiMatchQuery;
@@ -42,12 +45,14 @@ import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
import org.opensearch.client.opensearch.core.BulkRequest;
import org.opensearch.client.opensearch.core.BulkResponse;
import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
import org.opensearch.client.opensearch.core.DeleteByQueryResponse;
import org.opensearch.client.opensearch.core.DeleteRequest;
import org.opensearch.client.opensearch.core.IndexRequest;
import org.opensearch.client.opensearch.core.SearchRequest;
import org.opensearch.client.opensearch.core.SearchResponse;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
import org.opensearch.client.opensearch.core.bulk.DeleteOperation;
import org.opensearch.client.opensearch.core.bulk.IndexOperation;
import org.opensearch.client.opensearch.core.search.Highlight;
import org.opensearch.client.opensearch.core.search.HighlightField;
@@ -104,17 +109,33 @@ public class QuickSearchOpenSearchClient implements Closeable
String username = config.getOpensearchUsername();
String password = config.getOpensearchPassword();
BasicCredentialsProvider credentialsProvider = null;
if(username != null && password != null)
{
BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(
new AuthScope(httpHost),
new UsernamePasswordCredentials(username, password.toCharArray()));
builder.setHttpClientConfigCallback(httpClientBuilder ->
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
}
BasicCredentialsProvider finalCredentialsProvider = credentialsProvider;
builder.setHttpClientConfigCallback(httpClientBuilder ->
{
///////////////////////////////////////////////////////////////////////
// httpclient5 5.6+ decompresses gzip responses itself but leaves //
// the Content-Encoding header, so the OpenSearch transport would //
// gunzip the body a second time and fail. Let the transport alone //
// handle compression. //
///////////////////////////////////////////////////////////////////////
httpClientBuilder.disableContentCompression();
if(finalCredentialsProvider != null)
{
httpClientBuilder.setDefaultCredentialsProvider(finalCredentialsProvider);
}
return (httpClientBuilder);
});
transport = builder.build();
client = new OpenSearchClient(transport);
indexName = config.getOpensearchIndexName();
@@ -246,28 +267,7 @@ public class QuickSearchOpenSearchClient implements Closeable
.document(doc)))));
}
BulkRequest bulkRequest = BulkRequest.of(r -> r.operations(operations));
try
{
BulkResponse response = client.bulk(bulkRequest);
for(BulkResponseItem item : response.items())
{
if(item.error() != null)
{
result.addFailure(item.id() + ": " + item.error().reason());
}
else
{
result.addSuccess();
}
}
}
catch(IOException e)
{
throw new QException("Bulk index request failed: " + e.getMessage(), e);
}
executeBulk(operations, result, "index");
}
return (result);
@@ -275,6 +275,85 @@ public class QuickSearchOpenSearchClient implements Closeable
/*******************************************************************************
** Bulk-delete documents by their composite document IDs, in batches of at
** most batchSize.
**
** Every document is attempted; one failing item does not stop the others.
** A document that is not in the index counts as a success, because it is
** already absent. Per-item errors are recorded via addFailure().
**
** @param documentIds the composite document IDs; null is treated as empty
** @param batchSize maximum number of documents per bulk request
** @return accumulated result across all batches
** @throws QException if any batch request fails at the transport level
*******************************************************************************/
public BulkIndexResult deleteDocuments(List<String> documentIds, int batchSize) throws QException
{
BulkIndexResult result = new BulkIndexResult();
if(documentIds == null || documentIds.isEmpty())
{
return (result);
}
int total = documentIds.size();
for(int start = 0; start < total; start += batchSize)
{
List<String> batch = documentIds.subList(start, Math.min(start + batchSize, total));
List<BulkOperation> operations = new ArrayList<>();
for(String documentId : batch)
{
operations.add(BulkOperation.of(o -> o
.delete(DeleteOperation.of(d -> d
.index(indexName)
.id(documentId)))));
}
executeBulk(operations, result, "delete");
}
return (result);
}
/*******************************************************************************
** Send one bulk request and record each item's outcome in result.
**
** @param operations the bulk operations to send
** @param result accumulator for per-item successes and failures
** @param operationName used in the transport-failure message
** @throws QException if the request fails at the transport level
*******************************************************************************/
private void executeBulk(List<BulkOperation> operations, BulkIndexResult result, String operationName) throws QException
{
try
{
BulkResponse response = client.bulk(BulkRequest.of(r -> r.operations(operations)));
for(BulkResponseItem item : response.items())
{
if(item.error() != null)
{
result.addFailure(item.id() + ": " + item.error().reason());
}
else
{
result.addSuccess();
}
}
}
catch(IOException e)
{
throw new QException("Bulk " + operationName + " request failed: " + e.getMessage(), e);
}
}
/*******************************************************************************
** Delete a single document identified by sourceTable and recordId.
**
@@ -284,7 +363,7 @@ public class QuickSearchOpenSearchClient implements Closeable
*******************************************************************************/
public void deleteDocument(String sourceTable, String recordId) throws QException
{
String documentId = sourceTable + ":" + recordId;
String documentId = OpenSearchDocument.buildDocumentId(sourceTable, recordId);
try
{
@@ -325,6 +404,48 @@ public class QuickSearchOpenSearchClient implements Closeable
/*******************************************************************************
** Delete a table's documents that were not (re-)indexed at or after cutoff,
** including any that have no indexedAt value.
**
** Only documents visible to search are considered, so refresh the index
** first. Documents changed while the delete runs are skipped (version
** conflicts proceed), since they were just re-indexed.
**
** @param sourceTable the table whose documents to consider
** @param cutoff documents indexed before this instant are deleted
** @return the number of documents deleted
** @throws QException if the OpenSearch call fails
*******************************************************************************/
public Long deleteDocumentsIndexedBefore(String sourceTable, Instant cutoff) throws QException
{
try
{
DeleteByQueryResponse response = client.deleteByQuery(DeleteByQueryRequest.of(r -> r
.index(indexName)
.conflicts(Conflicts.Proceed)
.refresh(true)
.query(Query.of(q -> q
.bool(b -> b
.filter(f -> f
.term(t -> t
.field("sourceTable")
.value(FieldValue.of(sourceTable))))
.mustNot(mn -> mn
.range(rg -> rg
.field("indexedAt")
.gte(JsonData.of(cutoff.toString())))))))));
return (response.deleted() == null ? 0L : response.deleted());
}
catch(IOException e)
{
throw new QException("Failed to delete stale documents for table [" + sourceTable + "]: " + e.getMessage(), e);
}
}
/*******************************************************************************
** Execute a boosted multi-match search across the index.
**
src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java
+280
View File
@@ -0,0 +1,280 @@
/*
* Copyright 2024 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.quicksearch.processes;
import java.io.Serializable;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
/*******************************************************************************
** Process step that reconciles the search index with its source tables.
**
** For each table (or only the one named by the optional "tableName" input):
** 1. Re-index every source record, paging in primary-key order so each record
** is read exactly once, even while other rows are inserted or deleted.
** 2. If every document indexed, refresh the index and delete the table's
** documents that this run did not re-index (indexedAt before the run
** started). Those have no source record: their deletes were missed.
**
** Unlike FullReindexStep, the table's documents are never wiped first, so
** search keeps working during the run. When any document fails to index, the
** stale-document delete is skipped (it would remove that record's old
** document), and the run is marked FAILED.
**
** The cutoff comes from this JVM's clock, as do real-time indexedAt values on
** this node; application nodes are assumed to keep their clocks in sync.
*******************************************************************************/
public class ReconcileIndexStep extends AbstractIndexingStep
{
private static final QLogger LOG = QLogger.getLogger(ReconcileIndexStep.class);
public static final String RUN_TYPE = "RECONCILE";
/*******************************************************************************
** Totals from reconciling one table.
*******************************************************************************/
private record ReconcileCounts(Integer recordsIndexed, Long documentsRemoved)
{
}
/*******************************************************************************
** Reconcile every discovered table, or only the one named by "tableName".
**
** Output values: recordsIndexed (Integer) and documentsRemoved (Long),
** summed across tables.
**
** @throws QException if reconciling a table fails (its run is marked FAILED)
*******************************************************************************/
@Override
public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
{
String tableNameFilter = input.getValueString("tableName");
Integer recordsIndexed = 0;
Long documentsRemoved = 0L;
for(QuickSearchableTableConfig tableConfig : getDiscoveredTables())
{
if(tableNameFilter != null && !tableNameFilter.isBlank() && !tableNameFilter.equals(tableConfig.getTableName()))
{
continue;
}
ensureIndexRowExists(tableConfig.getTableName(), tableConfig);
ReconcileCounts counts = reconcileTable(tableConfig);
recordsIndexed = recordsIndexed + counts.recordsIndexed();
documentsRemoved = documentsRemoved + counts.documentsRemoved();
}
output.addValue("recordsIndexed", recordsIndexed);
output.addValue("documentsRemoved", documentsRemoved);
}
/*******************************************************************************
** Reconcile one table, recording the run in quickSearchIndexRun.
*******************************************************************************/
private ReconcileCounts reconcileTable(QuickSearchableTableConfig tableConfig) throws QException
{
String tableName = tableConfig.getTableName();
String primaryKeyField = tableConfig.getPrimaryKeyField();
QuickSearchOpenSearchClient client = getClient();
QuickSearchQBitConfig config = getConfig();
Integer indexId = queryIndexId(tableName);
QuickSearchIndexRun run = createRunRecord(indexId, RUN_TYPE);
Integer totalProcessed = 0;
Integer totalIndexed = 0;
Integer totalErrors = 0;
String firstError = null;
try
{
///////////////////////////////////////////////////////////////////////
// every document this run writes has indexedAt at or after this //
// instant, so afterwards any older document has no source record //
///////////////////////////////////////////////////////////////////////
Instant reconcileStart = Instant.now();
Serializable lastPrimaryKey = null;
while(true)
{
List<QRecord> batch = querySourceBatchAfter(tableName, primaryKeyField, lastPrimaryKey, config.getSourceBatchSize());
if(batch.isEmpty())
{
break;
}
List<OpenSearchDocument> documents = new ArrayList<>();
for(QRecord record : batch)
{
OpenSearchDocument document = IndexingUtils.buildDocument(record, tableConfig);
if(document != null)
{
documents.add(document);
}
}
if(!documents.isEmpty())
{
BulkIndexResult result = client.indexDocuments(documents, config.getBulkBatchSize());
totalIndexed = totalIndexed + result.getSuccessCount();
totalErrors = totalErrors + result.getFailureCount();
if(firstError == null && !result.getErrors().isEmpty())
{
firstError = result.getErrors().get(0);
}
}
totalProcessed = totalProcessed + batch.size();
lastPrimaryKey = batch.get(batch.size() - 1).getValue(primaryKeyField);
if(lastPrimaryKey == null)
{
//////////////////////////////////////////////////////////////
// paging needs a key to continue after; without one, the //
// next query would start over and never finish //
//////////////////////////////////////////////////////////////
throw new QException("Source record has no value in primary key field [" + primaryKeyField + "]");
}
}
Long documentsRemoved = 0L;
if(totalErrors.equals(0))
{
client.refreshIndex();
documentsRemoved = client.deleteDocumentsIndexedBefore(tableName, reconcileStart);
updateIndexRow(indexId, reconcileStart, totalProcessed);
}
else
{
LOG.warn("Reconcile had indexing errors; stale documents were not removed", logPair("tableName", tableName), logPair("errorCount", totalErrors));
}
String finalStatus = totalErrors.equals(0) ? "COMPLETED" : "FAILED";
completeRunRecord(run, finalStatus, totalProcessed, totalIndexed, totalErrors, firstError);
LOG.info("Reconcile complete", logPair("tableName", tableName), logPair("totalProcessed", totalProcessed),
logPair("totalIndexed", totalIndexed), logPair("totalErrors", totalErrors), logPair("documentsRemoved", documentsRemoved));
return (new ReconcileCounts(totalIndexed, documentsRemoved));
}
catch(Exception e)
{
LOG.warn("Reconcile failed", e, logPair("tableName", tableName));
completeRunRecord(run, "FAILED", totalProcessed, totalIndexed, totalErrors, e.getMessage());
throw new QException("Reconcile failed for table [" + tableName + "]: " + e.getMessage(), e);
}
}
/*******************************************************************************
** Query the next batch of source records, in primary-key order, after the
** given key (or from the start when it is null).
*******************************************************************************/
private List<QRecord> querySourceBatchAfter(String tableName, String primaryKeyField, Serializable afterPrimaryKey, Integer limit) throws QException
{
QQueryFilter filter = new QQueryFilter()
.withOrderBy(new QFilterOrderBy(primaryKeyField, true))
.withLimit(limit);
if(afterPrimaryKey != null)
{
filter.withCriteria(new QFilterCriteria(primaryKeyField, QCriteriaOperator.GREATER_THAN, afterPrimaryKey));
}
QueryInput queryInput = new QueryInput();
queryInput.setTableName(tableName);
queryInput.setFilter(filter);
return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()));
}
/*******************************************************************************
** Look up the ID of the quickSearchIndex row for a table.
*******************************************************************************/
private Integer queryIndexId(String tableName) throws QException
{
QueryInput queryInput = new QueryInput();
queryInput.setTableName(getConfig().getQuickSearchIndexTableName());
queryInput.setFilter(new QQueryFilter()
.withCriteria(new QFilterCriteria("tableName", QCriteriaOperator.EQUALS, tableName)));
List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
if(CollectionUtils.nullSafeIsEmpty(rows))
{
throw new QException("No QuickSearchIndex row found for table: " + tableName);
}
return (rows.get(0).getValueInteger("id"));
}
/*******************************************************************************
** Record a successful reconcile on the quickSearchIndex row. Records changed
** after reconcileStart may not have been read, so basepull resumes from there.
*******************************************************************************/
private void updateIndexRow(Integer indexId, Instant reconcileStart, Integer recordCount) throws QException
{
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(getConfig().getQuickSearchIndexTableName());
updateInput.setRecords(List.of(new QRecord()
.withValue("id", indexId)
.withValue("lastFullReindexTime", Instant.now())
.withValue("lastBasepullTime", reconcileStart)
.withValue("recordCount", recordCount)));
new UpdateAction().execute(updateInput);
}
}
src/main/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisher.java
+33 -5
View File
@@ -24,6 +24,7 @@ import java.util.Map;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import com.kingsrook.qbits.quicksearch.processes.IndexingUtils;
@@ -35,13 +36,15 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
** immediately against the OpenSearch client.
**
** Index events are grouped by table, converted to OpenSearchDocument instances
** using IndexingUtils, and bulk-indexed. Delete events are processed
** individually.
** using IndexingUtils, and bulk-indexed. Delete events are bulk-deleted.
** Either way, per-document failures are raised as a QException.
*******************************************************************************/
public class SynchronousIndexEventPublisher implements IndexEventPublisher
{
private static final QLogger LOG = QLogger.getLogger(SynchronousIndexEventPublisher.class);
private static final Integer MAX_ERRORS_IN_MESSAGE = 10;
private final QuickSearchOpenSearchClient client;
private final Map<String, QuickSearchableTableConfig> tableConfigsByName;
private final Integer bulkBatchSize;
@@ -119,14 +122,16 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
if(!allDocuments.isEmpty())
{
client.indexDocuments(allDocuments, bulkBatchSize);
BulkIndexResult result = client.indexDocuments(allDocuments, bulkBatchSize);
throwIfAnyFailed(result, "index", allDocuments.size());
}
}
/*******************************************************************************
** Publish delete events synchronously by calling deleteDocument for each event.
** Publish delete events synchronously as one bulk delete, so every document
** is attempted even when some fail. Throws if any document failed.
*******************************************************************************/
@Override
public void publishDeleteEvents(List<IndexEvent> events) throws QException
@@ -136,10 +141,33 @@ public class SynchronousIndexEventPublisher implements IndexEventPublisher
return;
}
List<String> documentIds = new ArrayList<>();
for(IndexEvent event : events)
{
client.deleteDocument(event.getTableName(), event.getRecordId());
documentIds.add(OpenSearchDocument.buildDocumentId(event.getTableName(), event.getRecordId()));
}
BulkIndexResult result = client.deleteDocuments(documentIds, bulkBatchSize);
throwIfAnyFailed(result, "delete", documentIds.size());
}
/*******************************************************************************
** Throw a QException naming the failed documents, if any bulk item failed,
** so failures reach the caller instead of leaving the index silently stale.
*******************************************************************************/
private void throwIfAnyFailed(BulkIndexResult result, String operationName, Integer attemptedCount) throws QException
{
if(result == null || Boolean.TRUE.equals(result.isFullySuccessful()))
{
return;
}
List<String> errors = result.getErrors();
List<String> errorsShown = errors.subList(0, Math.min(errors.size(), MAX_ERRORS_IN_MESSAGE));
throw (new QException("Failed to " + operationName + " " + result.getFailureCount() + " of " + attemptedCount
+ " search documents: " + String.join("; ", errorsShown)));
}
src/test/java/com/kingsrook/qbits/quicksearch/BaseQuickSearchTest.java
+1 -1
View File
@@ -79,7 +79,7 @@ public class BaseQuickSearchTest
////////////////////////////////////////////////////
// Set up fully-anonymous authentication //
////////////////////////////////////////////////////
qInstance.setAuthentication(new QAuthenticationMetaData()
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
src/test/java/com/kingsrook/qbits/quicksearch/ConfigDrivenIndexingIntegrationTest.java
+4 -2
View File
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.extension.ExtendWith;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
@@ -61,7 +62,8 @@ import static org.assertj.core.api.Assertions.assertThat;
** records, runs FullReindexStep, and verifies search results including custom
** recordLabelFormat.
*******************************************************************************/
@Testcontainers(disabledWithoutDocker = true)
@ExtendWith(RequiresDockerCondition.class)
@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class ConfigDrivenIndexingIntegrationTest
{
@@ -98,7 +100,7 @@ class ConfigDrivenIndexingIntegrationTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
qInstance.setAuthentication(new QAuthenticationMetaData()
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
src/test/java/com/kingsrook/qbits/quicksearch/IndexDriftIntegrationTest.java
+283
View File
@@ -0,0 +1,283 @@
/*
* Copyright 2024 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.quicksearch;
import java.io.Serializable;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
import com.kingsrook.qbits.quicksearch.actions.QuickSearchAction;
import com.kingsrook.qbits.quicksearch.actions.QuickSearchInput;
import com.kingsrook.qbits.quicksearch.actions.QuickSearchResult;
import com.kingsrook.qbits.quicksearch.annotations.QuickSearchField;
import com.kingsrook.qbits.quicksearch.annotations.QuickSearchable;
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import com.kingsrook.qbits.quicksearch.processes.ReconcileIndexStep;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Integration tests, against a real OpenSearch, that real-time indexing keeps
** the index in step with the source table through updates and deletes, and
** that the reconcile process repairs an index that has drifted.
**
** Each test uses its own record IDs and search terms, so tests are independent.
*******************************************************************************/
@ExtendWith(RequiresDockerCondition.class)
@Testcontainers
class IndexDriftIntegrationTest
{
private static final String BACKEND_NAME = "driftMemoryBackend";
private static final String TABLE_NAME = "driftCustomer";
private static final String INDEX_NAME = "quick_search_drift_test";
@Container
static GenericContainer<?> opensearch = new GenericContainer<>("opensearchproject/opensearch:2.11.0")
.withExposedPorts(9200)
.withEnv("discovery.type", "single-node")
.withEnv("plugins.security.disabled", "true")
.withEnv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "Admin123!")
.waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200));
/*******************************************************************************
** Entity annotated for quick search indexing.
*******************************************************************************/
@QuickSearchable(tableName = TABLE_NAME)
static class DriftCustomer
{
@QuickSearchField(weight = 2)
private String name;
@QuickSearchField
private String city;
}
/*******************************************************************************
** Build the instance, produce the QBit with real-time indexing on, and
** initialize QContext.
*******************************************************************************/
@BeforeAll
static void setUpAll() throws Exception
{
MemoryRecordStore.getInstance().reset();
QInstance qInstance = new QInstance();
qInstance.addBackend(new QBackendMetaData()
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
qInstance.addTable(new QTableMetaData()
.withName(TABLE_NAME)
.withBackendName(BACKEND_NAME)
.withPrimaryKeyField("id")
.withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
.withField(new QFieldMetaData("name", QFieldType.STRING))
.withField(new QFieldMetaData("city", QFieldType.STRING)));
QuickSearchQBitConfig config = new QuickSearchQBitConfig()
.withBackendName(BACKEND_NAME)
.withOpensearchHost(opensearch.getHost())
.withOpensearchPort(opensearch.getMappedPort(9200))
.withOpensearchIndexName(INDEX_NAME)
.withSearchableEntityClasses(List.of(DriftCustomer.class))
.withEnableRealTimeIndexing(true);
new QuickSearchQBitProducer().withConfig(config).produce(qInstance);
QContext.init(qInstance, new QSession());
}
/*******************************************************************************
** Clean up QContext and QBit state after all tests complete.
*******************************************************************************/
@AfterAll
static void tearDownAll()
{
QContext.clear();
QuickSearchQBitContext.clear();
MemoryRecordStore.getInstance().reset();
}
/*******************************************************************************
** Test: a partial update (one field) keeps the other fields searchable.
*******************************************************************************/
@Test
void testPartialUpdate_unchangedFieldsStaySearchable() throws Exception
{
insert(new QRecord().withValue("id", 1).withValue("name", "Ada").withValue("city", "Lovelaceville"));
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
updateInput.setRecords(List.of(new QRecord().withValue("id", 1).withValue("name", "Adeline")));
new UpdateAction().execute(updateInput);
assertThat(searchRecordIds("lovelaceville"))
.as("the unchanged city must stay searchable after updating only the name")
.containsExactly("1");
assertThat(searchRecordIds("adeline")).containsExactly("1");
}
/*******************************************************************************
** Test: deleting records removes their documents, even when the batch also
** holds a record whose document is already missing from the index.
*******************************************************************************/
@Test
void testDelete_removesDocuments_evenWithMissingDocumentInBatch() throws Exception
{
insert(
new QRecord().withValue("id", 10).withValue("name", "Tenley").withValue("city", "Deleteton"),
new QRecord().withValue("id", 11).withValue("name", "Elvira").withValue("city", "Deleteton"),
new QRecord().withValue("id", 12).withValue("name", "Twila").withValue("city", "Deleteton"));
////////////////////////////////////////////////////////////////////
// simulate earlier drift: record 12's document is already gone //
////////////////////////////////////////////////////////////////////
client().deleteDocument(TABLE_NAME, "12");
DeleteInput deleteInput = new DeleteInput();
deleteInput.setTableName(TABLE_NAME);
deleteInput.setPrimaryKeys(List.<Serializable>of(12, 10));
new DeleteAction().execute(deleteInput);
assertThat(searchRecordIds("deleteton"))
.as("only the record that was not deleted should remain")
.containsExactly("11");
}
/*******************************************************************************
** Test: the reconcile process repairs each kind of drift from the source
** table: an orphan document (record gone), a missing document, and a stale
** document (record changed since it was indexed).
*******************************************************************************/
@Test
void testReconcile_repairsOrphanMissingAndStaleDocuments() throws Exception
{
insert(
new QRecord().withValue("id", 21).withValue("name", "Stella").withValue("city", "Freshford"),
new QRecord().withValue("id", 22).withValue("name", "Mona").withValue("city", "Missingham"));
////////////////////////////////////////////////////////////////////
// drift the index away from the source table three ways //
////////////////////////////////////////////////////////////////////
Instant anHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
client().indexDocuments(List.of(
new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("99")
.withSearchableText("Ghostly Phantomburg").withIndexedAt(anHourAgo).withFieldValues(Map.of()),
new OpenSearchDocument().withSourceTable(TABLE_NAME).withRecordId("21")
.withSearchableText("Stella Oldtown").withIndexedAt(anHourAgo).withFieldValues(Map.of())), 10);
client().deleteDocument(TABLE_NAME, "22");
assertThat(searchRecordIds("phantomburg")).containsExactly("99");
assertThat(searchRecordIds("oldtown")).containsExactly("21");
assertThat(searchRecordIds("missingham")).isEmpty();
RunBackendStepInput input = new RunBackendStepInput();
input.addValue("tableName", TABLE_NAME);
RunBackendStepOutput output = new RunBackendStepOutput();
new ReconcileIndexStep().run(input, output);
assertThat(searchRecordIds("phantomburg")).as("orphan document removed").isEmpty();
assertThat(searchRecordIds("oldtown")).as("stale document replaced").isEmpty();
assertThat(searchRecordIds("freshford")).as("stale document replaced").containsExactly("21");
assertThat(searchRecordIds("missingham")).as("missing document restored").containsExactly("22");
assertThat(output.getValue("documentsRemoved")).isEqualTo(1L);
}
/*******************************************************************************
** Insert records through InsertAction (so real-time indexing runs).
*******************************************************************************/
private void insert(QRecord... records) throws QException
{
InsertInput insertInput = new InsertInput();
insertInput.setTableName(TABLE_NAME);
insertInput.setRecords(List.of(records));
new InsertAction().execute(insertInput);
}
/*******************************************************************************
** Refresh the index, then return the record IDs matching the term.
*******************************************************************************/
private List<String> searchRecordIds(String term) throws QException
{
client().refreshIndex();
return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME))
.getResults().stream()
.map(QuickSearchResult::getRecordId)
.toList());
}
/*******************************************************************************
** The live OpenSearch client from the QBit context.
*******************************************************************************/
private QuickSearchOpenSearchClient client()
{
return (QuickSearchQBitContext.getClient());
}
}
src/test/java/com/kingsrook/qbits/quicksearch/OpenSearchIntegrationTest.java
+4 -2
View File
@@ -47,6 +47,7 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.extension.ExtendWith;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
@@ -61,7 +62,8 @@ import static org.assertj.core.api.Assertions.assertThat;
** Tests run in order: index creation, full reindex, search with results,
** pagination, and no-results search.
*******************************************************************************/
@Testcontainers(disabledWithoutDocker = true)
@ExtendWith(RequiresDockerCondition.class)
@Testcontainers
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class OpenSearchIntegrationTest
{
@@ -115,7 +117,7 @@ class OpenSearchIntegrationTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
qInstance.setAuthentication(new QAuthenticationMetaData()
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitProducerTest.java
+3 -1
View File
@@ -162,6 +162,7 @@ class QuickSearchQBitProducerTest
assertThat(qInstance.getTable("myApp_quickSearchIndexRun")).isNotNull();
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess("myApp_" + QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
}
@@ -342,6 +343,7 @@ class QuickSearchQBitProducerTest
assertThat(qInstance.getProcess(QuickSearchQBitProducer.BASEPULL_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess(QuickSearchQBitProducer.FULL_REINDEX_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess(QuickSearchQBitProducer.RECONCILE_PROCESS_NAME)).isNotNull();
}
@@ -794,7 +796,7 @@ class QuickSearchQBitProducerTest
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
qInstance.setAuthentication(new QAuthenticationMetaData()
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerCondition.java
+72
View File
@@ -0,0 +1,72 @@
/*
* Copyright 2024 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.quicksearch;
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
import org.junit.jupiter.api.extension.ExecutionCondition;
import org.junit.jupiter.api.extension.ExtensionContext;
import org.testcontainers.DockerClientFactory;
/*******************************************************************************
** JUnit condition for integration tests that need Docker.
**
** Without Docker, the tests are skipped on a developer machine, but fail when
** CI=true, so a CI build can never pass with its integration tests silently
** skipped. Register it ahead of @Testcontainers, so no container is started
** when the tests are disabled.
*******************************************************************************/
public class RequiresDockerCondition implements ExecutionCondition
{
/*******************************************************************************
** Evaluate against the local Docker environment and the CI env var.
*******************************************************************************/
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context)
{
return (evaluate(DockerClientFactory.instance().isDockerAvailable(), System.getenv("CI")));
}
/*******************************************************************************
** Decide whether Docker-backed tests run, skip, or fail.
**
** @param dockerAvailable whether a usable Docker environment was found
** @param ciValue value of the CI environment variable (may be null)
** @return enabled when Docker is available; disabled when it is not and CI
** is not "true"
** @throws IllegalStateException when Docker is unavailable and CI is "true"
*******************************************************************************/
static ConditionEvaluationResult evaluate(Boolean dockerAvailable, String ciValue)
{
if(Boolean.TRUE.equals(dockerAvailable))
{
return (ConditionEvaluationResult.enabled("Docker is available"));
}
if("true".equalsIgnoreCase(ciValue))
{
throw (new IllegalStateException("Docker is required for integration tests when CI=true, but no usable Docker environment was found"));
}
return (ConditionEvaluationResult.disabled("Docker is not available; skipping (with CI=true these tests fail instead)"));
}
}
src/test/java/com/kingsrook/qbits/quicksearch/RequiresDockerConditionTest.java
+75
View File
@@ -0,0 +1,75 @@
/*
* Copyright 2024 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.quicksearch;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/*******************************************************************************
** Tests for RequiresDockerCondition.
*******************************************************************************/
class RequiresDockerConditionTest
{
/*******************************************************************************
** Docker available: integration tests run, whether or not CI is set.
*******************************************************************************/
@Test
void testEvaluate_dockerAvailable_enabled()
{
ConditionEvaluationResult local = RequiresDockerCondition.evaluate(true, null);
ConditionEvaluationResult ci = RequiresDockerCondition.evaluate(true, "true");
assertThat(local.isDisabled()).isFalse();
assertThat(ci.isDisabled()).isFalse();
}
/*******************************************************************************
** No Docker outside CI: integration tests are skipped.
*******************************************************************************/
@Test
void testEvaluate_noDockerOutsideCi_disabled()
{
assertThat(RequiresDockerCondition.evaluate(false, null).isDisabled()).isTrue();
assertThat(RequiresDockerCondition.evaluate(false, "").isDisabled()).isTrue();
assertThat(RequiresDockerCondition.evaluate(false, "false").isDisabled()).isTrue();
}
/*******************************************************************************
** No Docker with CI=true: evaluation throws, so the tests fail instead of
** silently skipping.
*******************************************************************************/
@Test
void testEvaluate_noDockerInCi_throws()
{
assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "true"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("CI=true");
assertThatThrownBy(() -> RequiresDockerCondition.evaluate(false, "TRUE"))
.isInstanceOf(IllegalStateException.class);
}
}
src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostDeleteCustomizerTest.java
+46 -1
View File
@@ -29,6 +29,7 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -79,7 +80,7 @@ class QuickSearchPostDeleteCustomizerTest
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
QContext.init(qInstance, new QSession());
}
@@ -144,6 +145,50 @@ class QuickSearchPostDeleteCustomizerTest
/*******************************************************************************
** Test that records the delete rejected (with errors), or that carry no
** primary key, do not remove documents from the index.
*******************************************************************************/
@SuppressWarnings("unchecked")
@Test
void testPostDelete_recordsWithErrorsOrNoPrimaryKey_skipped() throws QException
{
QRecord deleted = new QRecord().withValue(PK_FIELD, 5);
QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
rejected.addError(new BadInputStatusMessage("still referenced"));
QRecord noKey = new QRecord().withValue("name", "orphan");
DeleteInput deleteInput = new DeleteInput();
deleteInput.setTableName(TABLE_NAME);
List<QRecord> result = new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(deleted, rejected, noKey));
assertThat(result).containsExactly(deleted, rejected, noKey);
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
verify(mockPublisher).publishDeleteEvents(captor.capture());
assertThat(captor.getValue()).extracting(IndexEvent::getRecordId).containsExactly("5");
}
/*******************************************************************************
** Test that when no record is eligible, nothing is published.
*******************************************************************************/
@Test
void testPostDelete_onlyRejectedRecords_nothingPublished() throws QException
{
QRecord rejected = new QRecord().withValue(PK_FIELD, 6);
rejected.addError(new BadInputStatusMessage("still referenced"));
DeleteInput deleteInput = new DeleteInput();
deleteInput.setTableName(TABLE_NAME);
new QuickSearchPostDeleteCustomizer().postDelete(deleteInput, List.of(rejected));
verify(mockPublisher, never()).publishDeleteEvents(anyList());
}
/*******************************************************************************
** Test that post-delete with empty records list returns input without calling
** publisher.
src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostInsertCustomizerTest.java
+1 -1
View File
@@ -79,7 +79,7 @@ class QuickSearchPostInsertCustomizerTest
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
QContext.init(qInstance, new QSession());
}
src/test/java/com/kingsrook/qbits/quicksearch/customizers/QuickSearchPostUpdateCustomizerTest.java
+149 -29
View File
@@ -19,17 +19,26 @@ package com.kingsrook.qbits.quicksearch.customizers;
import java.util.List;
import java.util.Optional;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
@@ -50,23 +59,29 @@ import static org.mockito.Mockito.verify;
/*******************************************************************************
** Tests for QuickSearchPostUpdateCustomizer.
**
** Uses a memory backend, because the customizer re-reads updated records from
** the source table before indexing them.
*******************************************************************************/
class QuickSearchPostUpdateCustomizerTest
{
private static final String TABLE_NAME = "orders";
private static final String PK_FIELD = "id";
private static final String BACKEND_NAME = "memoryBackend";
private static final String TABLE_NAME = "orders";
private static final String PK_FIELD = "id";
private IndexEventPublisher mockPublisher;
/*******************************************************************************
** Set up mock publisher in QuickSearchQBitContext and a QContext with a test
** QInstance before each test.
** Set up mock publisher in QuickSearchQBitContext and a QContext with a
** memory-backed test QInstance before each test.
*******************************************************************************/
@BeforeEach
void setUp()
{
MemoryRecordStore.getInstance().reset();
mockPublisher = mock(IndexEventPublisher.class);
QuickSearchQBitContext.setPublisher(mockPublisher);
QuickSearchQBitContext.setDiscoveredTables(List.of(
@@ -74,14 +89,19 @@ class QuickSearchPostUpdateCustomizerTest
));
QInstance qInstance = new QInstance();
qInstance.addBackend(new QBackendMetaData()
.withName(BACKEND_NAME)
.withBackendType(MemoryBackendModule.class));
qInstance.addTable(new QTableMetaData()
.withName(TABLE_NAME)
.withBackendName(BACKEND_NAME)
.withPrimaryKeyField(PK_FIELD)
.withField(new QFieldMetaData(PK_FIELD, QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
qInstance.addBackend(new com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData().withName("testBackend"));
qInstance.setAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.FULLY_ANONYMOUS));
qInstance.getTable(TABLE_NAME).withBackendName("testBackend");
.withField(new QFieldMetaData("name", QFieldType.STRING))
.withField(new QFieldMetaData("description", QFieldType.STRING)));
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
.withName("anonymous")
.withType(QAuthenticationType.FULLY_ANONYMOUS));
QContext.init(qInstance, new QSession());
}
@@ -95,6 +115,33 @@ class QuickSearchPostUpdateCustomizerTest
{
QuickSearchQBitContext.clear();
QContext.clear();
MemoryRecordStore.getInstance().reset();
}
/*******************************************************************************
** Insert full records into the source table.
*******************************************************************************/
private void insertSourceRecords(QRecord... records) throws QException
{
InsertInput insertInput = new InsertInput();
insertInput.setTableName(TABLE_NAME);
insertInput.setRecords(List.of(records));
new InsertAction().execute(insertInput);
}
/*******************************************************************************
** Capture the events passed to the publisher's publishIndexEvents.
*******************************************************************************/
@SuppressWarnings("unchecked")
private List<IndexEvent> captureIndexEvents() throws QException
{
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
verify(mockPublisher).publishIndexEvents(captor.capture());
return (captor.getValue());
}
@@ -111,36 +158,107 @@ class QuickSearchPostUpdateCustomizerTest
/*******************************************************************************
** Test that post-update with valid records builds correct INDEX events and
** calls publisher.publishIndexEvents.
** Test that post-update builds INDEX events from the stored source records,
** not from the (possibly sparse) records passed in by the update.
*******************************************************************************/
@SuppressWarnings("unchecked")
@Test
void testPostUpdate_withRecords_publishesIndexEvents() throws QException
void testPostUpdate_withRecords_indexesStoredSourceRecords() throws QException
{
QRecord record1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
QRecord record2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
insertSourceRecords(
new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated").withValue("description", "Engineer"),
new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated").withValue("description", "Designer"));
QRecord sparse1 = new QRecord().withValue(PK_FIELD, 10).withValue("name", "Alice Updated");
QRecord sparse2 = new QRecord().withValue(PK_FIELD, 20).withValue("name", "Bob Updated");
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
List<QRecord> result = customizer.postUpdate(updateInput, List.of(record1, record2), Optional.empty());
List<QRecord> result = customizer.postUpdate(updateInput, List.of(sparse1, sparse2), Optional.empty());
assertThat(result).containsExactly(record1, record2);
assertThat(result).containsExactly(sparse1, sparse2);
ArgumentCaptor<List<IndexEvent>> captor = ArgumentCaptor.forClass(List.class);
verify(mockPublisher).publishIndexEvents(captor.capture());
List<IndexEvent> events = captor.getValue();
List<IndexEvent> events = captureIndexEvents();
assertThat(events).hasSize(2);
assertThat(events.get(0).getAction()).isEqualTo(IndexEventAction.INDEX);
assertThat(events.get(0).getTableName()).isEqualTo(TABLE_NAME);
assertThat(events.get(0).getRecordId()).isEqualTo("10");
assertThat(events.get(0).getRecord()).isSameAs(record1);
assertThat(events.get(1).getAction()).isEqualTo(IndexEventAction.INDEX);
assertThat(events.get(1).getRecordId()).isEqualTo("20");
assertThat(events.get(1).getRecord()).isSameAs(record2);
assertThat(events).allSatisfy(e ->
{
assertThat(e.getAction()).isEqualTo(IndexEventAction.INDEX);
assertThat(e.getTableName()).isEqualTo(TABLE_NAME);
});
assertThat(events).extracting(IndexEvent::getRecordId).containsExactlyInAnyOrder("10", "20");
IndexEvent aliceEvent = events.stream().filter(e -> "10".equals(e.getRecordId())).findFirst().orElseThrow();
assertThat(aliceEvent.getRecord().getValueString("name")).isEqualTo("Alice Updated");
assertThat(aliceEvent.getRecord().getValueString("description")).isEqualTo("Engineer");
}
/*******************************************************************************
** End to end through UpdateAction: a partial update of one field must not drop
** the other indexed fields from the document.
*******************************************************************************/
@Test
void testUpdateAction_partialUpdate_reindexesFullRecord() throws QException
{
QContext.getQInstance().getTable(TABLE_NAME)
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(QuickSearchPostUpdateCustomizer.class));
insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice").withValue("description", "Engineer"));
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
updateInput.setRecords(List.of(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alicia")));
new UpdateAction().execute(updateInput);
List<IndexEvent> events = captureIndexEvents();
assertThat(events).hasSize(1);
assertThat(events.get(0).getRecordId()).isEqualTo("1");
assertThat(events.get(0).getRecord().getValueString("name")).isEqualTo("Alicia");
assertThat(events.get(0).getRecord().getValueString("description")).isEqualTo("Engineer");
}
/*******************************************************************************
** Test that records the update rejected (with errors) are not re-indexed.
*******************************************************************************/
@Test
void testPostUpdate_recordWithErrors_notIndexed() throws QException
{
insertSourceRecords(
new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good"),
new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected"));
QRecord good = new QRecord().withValue(PK_FIELD, 1).withValue("name", "Good");
QRecord rejected = new QRecord().withValue(PK_FIELD, 2).withValue("name", "Rejected");
rejected.addError(new BadInputStatusMessage("not allowed"));
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(good, rejected), Optional.empty());
List<IndexEvent> events = captureIndexEvents();
assertThat(events).extracting(IndexEvent::getRecordId).containsExactly("1");
}
/*******************************************************************************
** Test that when no updated record can be read back from the source table,
** nothing is published.
*******************************************************************************/
@Test
void testPostUpdate_recordNotInSource_nothingPublished() throws QException
{
UpdateInput updateInput = new UpdateInput();
updateInput.setTableName(TABLE_NAME);
new QuickSearchPostUpdateCustomizer().postUpdate(updateInput, List.of(new QRecord().withValue(PK_FIELD, 99)), Optional.empty());
verify(mockPublisher, never()).publishIndexEvents(anyList());
}
@@ -209,6 +327,7 @@ class QuickSearchPostUpdateCustomizerTest
@Test
void testPostUpdate_publisherThrows_returnsRecordsWithoutThrowing() throws QException
{
insertSourceRecords(new QRecord().withValue(PK_FIELD, 1).withValue("name", "Alice"));
doThrow(new QException("publish failed"))
.when(mockPublisher).publishIndexEvents(anyList());
@@ -218,6 +337,7 @@ class QuickSearchPostUpdateCustomizerTest
QuickSearchPostUpdateCustomizer customizer = new QuickSearchPostUpdateCustomizer();
assertThatNoException().isThrownBy(() -> customizer.postUpdate(updateInput, List.of(record), Optional.empty()));
verify(mockPublisher).publishIndexEvents(anyList());
}
}
src/test/java/com/kingsrook/qbits/quicksearch/opensearch/QuickSearchOpenSearchClientTest.java
+47
View File
@@ -261,4 +261,51 @@ class QuickSearchOpenSearchClientTest
}
}
/*******************************************************************************
** deleteDocuments with a null or empty list is a no-op, even with no server.
*******************************************************************************/
@Test
void testDeleteDocuments_emptyList_returnsEmptyResult() throws QException
{
QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
try
{
BulkIndexResult emptyResult = client.deleteDocuments(List.of(), 100);
BulkIndexResult nullResult = client.deleteDocuments(null, 100);
assertThat(emptyResult.getSuccessCount()).isEqualTo(0);
assertThat(emptyResult.getFailureCount()).isEqualTo(0);
assertThat(nullResult.getSuccessCount()).isEqualTo(0);
}
finally
{
client.close();
}
}
/*******************************************************************************
** deleteDocuments throws QException when the server is unavailable.
*******************************************************************************/
@Test
void testDeleteDocuments_serverUnavailable_throwsQException() throws QException
{
QuickSearchOpenSearchClient client = new QuickSearchOpenSearchClient(plainConfig());
try
{
assertThatThrownBy(() -> client.deleteDocuments(List.of("orders:1", "orders:2"), 100))
.isInstanceOf(QException.class)
.hasMessageContaining("Bulk delete");
}
finally
{
client.close();
}
}
}
src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java
+281
View File
@@ -0,0 +1,281 @@
/*
* Copyright 2024 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.quicksearch.processes;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/*******************************************************************************
** Tests for ReconcileIndexStep.
*******************************************************************************/
class ReconcileIndexStepTest extends BaseQuickSearchTest
{
private QuickSearchOpenSearchClient mockClient;
/*******************************************************************************
** Install a mock client that reports every document as indexed.
*******************************************************************************/
@BeforeEach
void setUpClient() throws QException
{
mockClient = mock(QuickSearchOpenSearchClient.class);
when(mockClient.indexDocuments(anyList(), anyInt())).thenAnswer(invocation ->
{
List<?> docs = invocation.getArgument(0);
return (new BulkIndexResult().withSuccessCount(docs.size()));
});
when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(0L);
QuickSearchQBitContext.setClient(mockClient);
}
/*******************************************************************************
** Insert count source records into the testEntity table.
*******************************************************************************/
private void insertTestEntities(int count) throws QException
{
List<QRecord> records = new ArrayList<>();
for(int i = 1; i <= count; i++)
{
records.add(new QRecord().withValue("name", "Entity " + i).withValue("description", "Description " + i));
}
InsertInput insertInput = new InsertInput();
insertInput.setTableName(TEST_ENTITY_TABLE);
insertInput.setRecords(records);
new InsertAction().execute(insertInput);
}
/*******************************************************************************
** Query all rows of the given table.
*******************************************************************************/
private List<QRecord> queryAll(String tableName) throws QException
{
QueryInput queryInput = new QueryInput();
queryInput.setTableName(tableName);
queryInput.setFilter(new QQueryFilter());
return (new QueryAction().execute(queryInput).getRecords());
}
/*******************************************************************************
** Collect the record IDs of every document passed to indexDocuments.
*******************************************************************************/
@SuppressWarnings("unchecked")
private List<String> indexedRecordIds() throws QException
{
ArgumentCaptor<List<OpenSearchDocument>> captor = ArgumentCaptor.forClass(List.class);
verify(mockClient, atLeastOnce()).indexDocuments(captor.capture(), anyInt());
List<String> recordIds = new ArrayList<>();
for(List<OpenSearchDocument> batch : captor.getAllValues())
{
batch.forEach(doc -> recordIds.add(doc.getRecordId()));
}
return (recordIds);
}
/*******************************************************************************
** Test: every source record is re-indexed, then documents not re-indexed by
** this run are removed, after a refresh. The table is never wiped first.
*******************************************************************************/
@Test
void testReconcile_reindexesSourceThenRemovesStaleDocuments() throws QException
{
insertTestEntities(3);
when(mockClient.deleteDocumentsIndexedBefore(anyString(), any())).thenReturn(2L);
Instant beforeRun = Instant.now();
RunBackendStepOutput output = new RunBackendStepOutput();
new ReconcileIndexStep().run(new RunBackendStepInput(), output);
Instant afterRun = Instant.now();
assertThat(indexedRecordIds()).containsExactlyInAnyOrder("1", "2", "3");
ArgumentCaptor<Instant> cutoffCaptor = ArgumentCaptor.forClass(Instant.class);
InOrder inOrder = inOrder(mockClient);
inOrder.verify(mockClient).indexDocuments(anyList(), anyInt());
inOrder.verify(mockClient).refreshIndex();
inOrder.verify(mockClient).deleteDocumentsIndexedBefore(eq(TEST_ENTITY_TABLE), cutoffCaptor.capture());
assertThat(cutoffCaptor.getValue()).isBetween(beforeRun, afterRun);
verify(mockClient, never()).deleteDocumentsForTable(anyString());
assertThat(output.getValueInteger("recordsIndexed")).isEqualTo(3);
assertThat(output.getValue("documentsRemoved")).isEqualTo(2L);
}
/*******************************************************************************
** Test: source records are paged by primary key, so every record is read
** exactly once across batches.
*******************************************************************************/
@Test
void testReconcile_pagesByPrimaryKey_eachRecordOnce() throws QException
{
QuickSearchQBitContext.getConfig().withSourceBatchSize(2);
insertTestEntities(5);
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
verify(mockClient, times(3)).indexDocuments(anyList(), anyInt());
assertThat(indexedRecordIds()).containsExactly("1", "2", "3", "4", "5");
}
/*******************************************************************************
** Test: when any document fails to index, stale documents are not removed
** (a document that failed to re-index would otherwise be purged), and the
** run is marked FAILED.
*******************************************************************************/
@Test
void testReconcile_indexingErrors_keepsDocumentsAndFailsRun() throws QException
{
insertTestEntities(2);
BulkIndexResult partial = new BulkIndexResult();
partial.addSuccess();
partial.addFailure("testEntity:2: mapper_parsing_exception");
when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(partial);
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
assertThat(runs).hasSize(1);
assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
assertThat(runs.get(0).getValueInteger("errorCount")).isEqualTo(1);
assertThat(runs.get(0).getValueString("errorMessage")).contains("mapper_parsing_exception");
}
/*******************************************************************************
** Test: a client exception fails the run record and is rethrown.
*******************************************************************************/
@Test
void testReconcile_clientThrows_failsRunAndThrows() throws QException
{
insertTestEntities(1);
when(mockClient.indexDocuments(anyList(), anyInt())).thenThrow(new QException("connection refused"));
assertThatThrownBy(() -> new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
.isInstanceOf(QException.class)
.hasMessageContaining("Reconcile failed");
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
assertThat(runs).hasSize(1);
assertThat(runs.get(0).getValueString("status")).isEqualTo("FAILED");
assertThat(runs.get(0).getValueString("errorMessage")).contains("connection refused");
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
}
/*******************************************************************************
** Test: a successful run completes its run record and updates the index row.
*******************************************************************************/
@Test
void testReconcile_success_completesRunAndUpdatesIndexRow() throws QException
{
insertTestEntities(4);
Instant beforeRun = Instant.now();
new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
List<QRecord> runs = queryAll(QuickSearchIndexRun.TABLE_NAME);
assertThat(runs).hasSize(1);
assertThat(runs.get(0).getValueString("runType")).isEqualTo("RECONCILE");
assertThat(runs.get(0).getValueString("status")).isEqualTo("COMPLETED");
assertThat(runs.get(0).getValueInteger("recordsProcessed")).isEqualTo(4);
assertThat(runs.get(0).getValueInteger("recordsIndexed")).isEqualTo(4);
List<QRecord> indexRows = queryAll(QuickSearchIndex.TABLE_NAME);
assertThat(indexRows).hasSize(1);
assertThat(indexRows.get(0).getValueInteger("recordCount")).isEqualTo(4);
assertThat(indexRows.get(0).getValueInstant("lastFullReindexTime")).isAfterOrEqualTo(beforeRun);
assertThat(indexRows.get(0).getValueInstant("lastBasepullTime")).isAfterOrEqualTo(beforeRun);
}
/*******************************************************************************
** Test: the tableName input limits the run to that table.
*******************************************************************************/
@Test
void testReconcile_unknownTableNameInput_nothingReconciled() throws QException
{
insertTestEntities(1);
RunBackendStepInput input = new RunBackendStepInput();
input.addValue("tableName", "someOtherTable");
new ReconcileIndexStep().run(input, new RunBackendStepOutput());
verify(mockClient, never()).indexDocuments(anyList(), anyInt());
verify(mockClient, never()).deleteDocumentsIndexedBefore(anyString(), any());
assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).isEmpty();
}
}
src/test/java/com/kingsrook/qbits/quicksearch/publisher/SynchronousIndexEventPublisherTest.java
+58 -5
View File
@@ -34,6 +34,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -167,10 +168,11 @@ class SynchronousIndexEventPublisherTest
/*******************************************************************************
** Test publishDeleteEvents calls client.deleteDocument for each event.
** Test publishDeleteEvents removes all documents in one bulk call, so one
** failing document cannot stop the others from being removed.
*******************************************************************************/
@Test
void testPublishDeleteEvents_callsClientForEachEvent() throws QException
void testPublishDeleteEvents_bulkDeletesAllDocuments() throws QException
{
List<IndexEvent> events = List.of(
new IndexEvent()
@@ -182,10 +184,61 @@ class SynchronousIndexEventPublisherTest
.withRecordId("20")
.withAction(IndexEventAction.DELETE));
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult());
publisher.publishDeleteEvents(events);
verify(mockClient).deleteDocument("orders", "10");
verify(mockClient).deleteDocument("orders", "20");
verify(mockClient).deleteDocuments(List.of("orders:10", "orders:20"), BATCH_SIZE);
verify(mockClient, never()).deleteDocument(anyString(), anyString());
}
/*******************************************************************************
** Test publishDeleteEvents throws, naming the failed documents, when some
** bulk items fail.
*******************************************************************************/
@Test
void testPublishDeleteEvents_itemFailures_throwsQException() throws QException
{
BulkIndexResult result = new BulkIndexResult();
result.addSuccess();
result.addFailure("orders:20: shard unavailable");
when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(result);
List<IndexEvent> events = List.of(
new IndexEvent().withTableName("orders").withRecordId("10").withAction(IndexEventAction.DELETE),
new IndexEvent().withTableName("orders").withRecordId("20").withAction(IndexEventAction.DELETE));
assertThatThrownBy(() -> publisher.publishDeleteEvents(events))
.isInstanceOf(QException.class)
.hasMessageContaining("1 of 2")
.hasMessageContaining("orders:20: shard unavailable");
}
/*******************************************************************************
** Test publishIndexEvents throws, naming the failed documents, when some
** bulk items fail (instead of silently dropping the failures).
*******************************************************************************/
@Test
void testPublishIndexEvents_itemFailures_throwsQException() throws QException
{
BulkIndexResult result = new BulkIndexResult();
result.addFailure("orders:1: mapper_parsing_exception");
when(mockClient.indexDocuments(anyList(), anyInt())).thenReturn(result);
IndexEvent event = new IndexEvent()
.withTableName("orders")
.withRecordId("1")
.withAction(IndexEventAction.INDEX)
.withRecord(new QRecord().withValue("id", 1).withValue("orderNumber", "ORD-001"));
assertThatThrownBy(() -> publisher.publishIndexEvents(List.of(event)))
.isInstanceOf(QException.class)
.hasMessageContaining("1 of 1")
.hasMessageContaining("orders:1: mapper_parsing_exception");
}
@@ -209,7 +262,7 @@ class SynchronousIndexEventPublisherTest
void testPublishDeleteEvents_clientThrows_propagatesException() throws QException
{
doThrow(new QException("delete failed"))
.when(mockClient).deleteDocument("orders", "1");
.when(mockClient).deleteDocuments(anyList(), anyInt());
IndexEvent event = new IndexEvent()
.withTableName("orders")