From f30857fb999ded7b01ec5384b7efaf4ccb02a15a Mon Sep 17 00:00:00 2001 From: Hyeoncheol Kim Date: Tue, 21 Jul 2026 21:14:08 +0900 Subject: [PATCH] feat: expand MongoDB analytics operator parity --- CHANGELOG.md | 16 +- README.md | 2 +- build.gradle.kts | 3 +- docs/COMPATIBILITY.md | 26 +- docs/RELEASE_CHECKLIST.md | 7 +- docs/ROADMAP.md | 3 +- docs/SUPPORT_MATRIX.md | 8 +- jongodb-spring-suite/build.gradle.kts | 1 + jongodb-testkit/build.gradle.kts | 1 + .../org/jongodb/command/CommandStore.java | 35 +- .../command/EngineBackedCommandStore.java | 16 +- .../command/FindAndModifyCommandHandler.java | 28 +- .../FindOneAndUpdateCommandHandler.java | 31 +- .../command/UpdateArrayFiltersSubset.java | 10 + .../jongodb/command/UpdateCommandHandler.java | 37 +- .../jongodb/command/UpdatePipelineSubset.java | 161 +---- .../engine/AggregationExpressions.java | 603 +++++++++++++++++ .../jongodb/engine/AggregationPipeline.java | 628 ++++++++++++++---- .../org/jongodb/engine/CollectionStore.java | 8 + .../engine/InMemoryCollectionStore.java | 97 +++ .../jongodb/engine/MongoValueComparator.java | 305 +++++++++ .../org/jongodb/engine/UpdateApplier.java | 68 +- .../command/CommandDispatcherE2ETest.java | 108 ++- .../engine/AggregationPipelineTest.java | 109 +++ .../org/jongodb/engine/UpdateApplierTest.java | 52 +- 25 files changed, 2051 insertions(+), 312 deletions(-) create mode 100644 src/main/java/org/jongodb/engine/AggregationExpressions.java create mode 100644 src/main/java/org/jongodb/engine/MongoValueComparator.java diff --git a/CHANGELOG.md b/CHANGELOG.md index bb729b0..9079d46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,20 @@ All notable changes to this project are documented in this file. ## [Unreleased] +### Changed +- Default local project version moved to `0.1.10-SNAPSHOT`. + +## [0.1.10] - 2026-07-21 + +### Added +- Added MongoDB-style `$min`/`$max` update operators with BSON ordering, mixed numeric types, dates, dotted paths, upserts, and modifier conflict detection. +- Added expression-based aggregation-pipeline updates for `$addFields`/`$set`, `$project`/`$unset`, and `$replaceRoot`/`$replaceWith`. +- Added aggregation expressions for conditional/comparison/type/null handling, min/max, basic arithmetic, `$mergeObjects`, and UTC `$dateTrunc` with calendar units and `binSize`. +- Expanded `$group` with core, set/object merge, percentile/median, and first/last/min/max/top/bottom N accumulators. +- Added `$setWindowFields` partition/sort support with `$shift`, `$documentNumber`, `$rank`, and `$denseRank`. + +## [0.1.9] - 2026-05-09 + ### Added - Added `$unwind.includeArrayIndex` support, including MongoDB-compatible `null` index handling for scalar and preserved null/missing/empty inputs. - Added `listCollections`, `drop`, and `dropDatabase` command handlers for metadata lookup and fixture cleanup flows. @@ -54,7 +68,7 @@ All notable changes to this project are documented in this file. - Fixed upsert seed extraction to honor equality clauses nested inside `$and`, restoring duplicate-key behavior for `findOneAndUpdate` lock-style filters. ### Changed -- Default local project version moved to `0.1.9-SNAPSHOT`. +- Default local project version moved to `0.1.9-SNAPSHOT` before the `v0.1.9` release. ## [0.1.3] - 2026-02-24 diff --git a/README.md b/README.md index 60fe8f2..51ff49e 100644 --- a/README.md +++ b/README.md @@ -176,7 +176,7 @@ This project targets integration-test compatibility for common Spring data paths | --- | --- | --- | | Command surface | 25 handlers | Mix of `Supported` and `Partial` | | Query language | Core comparison/logical/array/regex + partial `$expr` (including `$add` subset) | Advanced parity incomplete | -| Aggregation | Core stages, `$unwind.includeArrayIndex`, selected Tier-2 stages + minimal `$graphLookup` subset | Full operator coverage not implemented | +| Aggregation | Core stages, analytics accumulators, UTC `$dateTrunc`, `$setWindowFields` sequence subset, and minimal `$graphLookup` | Full operator coverage not implemented | | Transactions | Single-process session/transaction flow | Namespace-aware commit merge + snapshot reads (`find`/`aggregate`/`countDocuments`) + deterministic retry labels/contracts | | Deployment profile | Standalone + single-node replica-set semantic profile | Replica-set profile exposes primary-only handshake/URI semantics for driver compatibility | | Wire protocol | `OP_MSG` + `OP_QUERY` | In-process ingress and standalone TCP launcher mode, with OP_QUERY namespace-based `$db` fallback | diff --git a/build.gradle.kts b/build.gradle.kts index 9810f16..eb3e10d 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -8,7 +8,7 @@ plugins { } group = providers.gradleProperty("publishGroup").orElse("io.github.midagedev").get() -version = providers.gradleProperty("publishVersion").orElse("0.1.9-SNAPSHOT").get() +version = providers.gradleProperty("publishVersion").orElse("0.1.10-SNAPSHOT").get() repositories { mavenCentral() @@ -42,6 +42,7 @@ dependencies { testImplementation(platform("org.junit:junit-bom:5.10.2")) testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") testImplementation("org.mongodb:mongodb-driver-sync:4.11.2") testImplementation("com.tngtech.archunit:archunit-junit5:1.3.0") testImplementation("org.springframework:spring-context:6.1.17") diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index d4234d2..b531ba3 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -1,6 +1,6 @@ # Compatibility Matrix -Status date: 2026-05-09 +Status date: 2026-07-21 This page describes implemented behavior in this repository. It is a code-level matrix, not a MongoDB claim. @@ -28,7 +28,7 @@ Certification context: | `listCollections` | Partial | Cursor-shaped collection metadata subset for fixture discovery | | `drop` | Partial | Collection cleanup subset with deterministic `NamespaceNotFound` error | | `dropDatabase` | Partial | Database-scoped cleanup subset | -| `update` | Partial | Operator set intentionally limited; update pipeline subset supports `$set`/`$unset` stages without expression evaluation | +| `update` | Partial | Supports `$min`/`$max` and core modifiers; aggregation-pipeline updates support MongoDB's update-stage allowlist with the expression subset below | | `delete` | Supported | `limit` 0/1 behavior | | `bulkWrite` | Partial | Ordered mode only (`ordered=true`); supports `insertOne/updateOne/updateMany/deleteOne/deleteMany/replaceOne` and stops on first write error | | `clientBulkWrite` | Partial | UTF importer subset rewrites ordered single-namespace models to `bulkWrite`; mixed namespaces, `ordered=false`, and `verboseResults=true` are deterministic unsupported paths | @@ -36,7 +36,7 @@ Certification context: | `countDocuments` | Partial | Filter + skip/limit + hint/readConcern; collation subset applied to filter comparison | | `runCommand` | Partial | UTF importer subset supports `ping`, `buildInfo`, `listIndexes`, `listCollections`, `count`, `distinct`, `drop`, `dropDatabase`; other command names fail with deterministic unsupported reasons | | `replaceOne` | Partial | Rewrites to single replacement `update` path (`multi=false`) | -| `findOneAndUpdate` | Partial | Rewrites to `findAndModify`; supports operator updates plus update-pipeline subset (`$set`/`$unset`, no expression evaluation), `arrayFilters` subset, and projection include/exclude subset (including `_id` override) | +| `findOneAndUpdate` | Partial | Rewrites to `findAndModify`; supports operator and expression-based pipeline updates, `arrayFilters` subset, and projection include/exclude subset (including `_id` override) | | `findOneAndReplace` | Partial | Rewrites to `findAndModify`; replacement updates only; supports projection include/exclude subset (including `_id` override) | | `commitTransaction`, `abortTransaction` | Supported | Session/txn envelope supported | @@ -67,7 +67,7 @@ Not implemented: Implemented stages: - `$match` - `$project` -- `$group` (subset: `$sum`, `$first`, `$addToSet`) +- `$group` with `$sum`, `$avg`, `$min`, `$max`, `$first`, `$last`, `$push`, `$addToSet`, `$mergeObjects`, `$percentile`, `$median`, `$firstN`, `$lastN`, `$minN`, `$maxN`, `$topN`, and `$bottomN` - `$sort` - `$limit`, `$skip` - `$unwind` (including `includeArrayIndex`) @@ -82,13 +82,16 @@ Implemented stages: - `$lookup` (local/foreign and pipeline+let subset) - `$unionWith` - `$graphLookup` (minimal subset: `from`, `startWith`, `connectFromField`, `connectToField`, `as`, optional `maxDepth`) +- `$setWindowFields` subset with `partitionBy`, `sortBy`, `$shift`, `$documentNumber`, `$rank`, and `$denseRank` - `$out` (terminal string-target subset: replaces target collection contents and returns empty result set) - `$merge` (terminal string-target or `{into: }` subset; merges by `_id`) Not implemented or partial: - unsupported stages return deterministic fail-fast -- many advanced expression operators are still missing -- `$group` accumulators other than `$sum` are not available +- expression support includes field references, `$literal`, `$cond`, `$ifNull`, `$type`, comparisons, logical operators, `$min`/`$max`, basic arithmetic, `$mergeObjects`, and `$dateTrunc` +- `$dateTrunc` supports second through year units, `binSize`, `startOfWeek`, and UTC timezone aliases; non-UTC timezones fail explicitly +- `$percentile`/`$median` accept `method: "approximate"` and use deterministic exact interpolation internally +- window accumulators and range/time windows beyond the listed `$setWindowFields` subset are not implemented - `$merge` options beyond the terminal string / `{into: }` subset are deterministic unsupported paths - `$graphLookup` options outside current subset (for example `depthField`, `restrictSearchWithMatch`) are deterministic unsupported paths - `bypassDocumentValidation` for aggregate is excluded from current differential corpus @@ -96,9 +99,10 @@ Not implemented or partial: ## Update Semantics Supported: -- operator updates: `$set`, `$inc`, `$unset` +- operator updates: `$set`, `$setOnInsert`, `$inc`, `$min`, `$max`, `$unset`, `$addToSet` - `arrayFilters` subset for `$set`/`$unset` paths using `$[identifier]` bindings -- update pipeline subset: `$set`/`$unset` stages with literal values +- update pipeline stages: `$addFields`/`$set`, `$project`/`$unset`, and `$replaceRoot`/`$replaceWith` +- update pipeline expressions: field references, `$literal`, `$cond`, `$ifNull`, `$type`, comparisons, logical operators, `$min`/`$max`, basic arithmetic, `$mergeObjects`, and `$dateTrunc` - replacement updates (with `multi=false`) - upsert for operator and replacement forms - same update constraints apply to `bulkWrite` update/replace operations @@ -107,8 +111,8 @@ Not supported: - advanced `arrayFilters` forms (unsupported operators, missing bindings, unsupported path/operator combinations) - positional updates (`$`, `$[]`) - update operators outside the supported set -- update pipeline stages outside `$set`/`$unset` -- update pipeline expressions (field references/operator expressions inside stage values) +- update pipeline stages outside MongoDB's update-stage allowlist +- aggregation expressions outside the listed subset - replacement updates with `multi=true` ## R3 Query/Update Corpus Exclusions @@ -122,7 +126,7 @@ differential parity counts: - dot/dollar insert payload forms outside the deterministic insert subset; dollar-prefixed subfields under `_id` fail with `code=52` - `runCommand` command names outside the imported subset (`ping`, `buildInfo`, `listIndexes`, `listCollections`, `count`, `distinct`, `drop`, `dropDatabase`) - update operations using unsupported `arrayFilters` forms (outside `$set`/`$unset` subset) -- update pipeline forms outside the supported subset (`$set`/`$unset` stages with literal values) +- update pipeline forms outside the supported stage/expression subsets listed above - replacement updates requested with `multi=true` ## UTF Import Profiles diff --git a/docs/RELEASE_CHECKLIST.md b/docs/RELEASE_CHECKLIST.md index 62e9055..95919ed 100644 --- a/docs/RELEASE_CHECKLIST.md +++ b/docs/RELEASE_CHECKLIST.md @@ -1,6 +1,6 @@ # Release Checklist -Status date: 2026-05-09 +Status date: 2026-07-21 ## R3 Certification Sign-Off @@ -25,10 +25,10 @@ Status date: 2026-05-09 ## Current Release-Line Notes -- Latest Java release tag: `v0.1.8` (run `23785415137`, commit `7b8ef1f`). +- Latest Java release tag: `v0.1.9` (run `25603328584`, commit `46ddb9b`). - Latest Node adapter tag: `node-v0.1.4` (run `22379340586`, commit `868bbc7`). - Latest compatibility certification snapshot: commit `b44be73` (runs `25603109902`, `25603109886`, `25603109905`, `25603109900`). -- Next Java tag candidate: `v0.1.9`; regenerate certification artifacts against the release-candidate commit, not reuse historical tag evidence. +- Next Java tag candidate: `v0.1.10`; regenerate certification artifacts against the release-candidate commit, not reuse historical tag evidence. ## Tagging Gate @@ -77,6 +77,7 @@ Use this when validating `@jongodb/memory-server` canary automation: | Version | Date (UTC) | Commit | Maven | GitHub Actions | | --- | --- | --- | --- | --- | +| `0.1.9` | `2026-05-09` | `46ddb9b` | `io.github.midagedev:jongodb:0.1.9` | run `25603328584` | | `0.1.8` | `2026-03-31` | `7b8ef1f` | `io.github.midagedev:jongodb:0.1.8` | run `23785415137` | | `0.1.7` | `2026-03-05` | `b7c4a13` | `io.github.midagedev:jongodb:0.1.7` | run `22723048147` | | `0.1.6` | `2026-03-01` | `12a259c` | `io.github.midagedev:jongodb:0.1.6` | run `22541258962` | diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 0525968..b93ac13 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -49,6 +49,7 @@ Completed in current wave: - `listCollections`, `drop`, and `dropDatabase` command subsets added for fixture discovery/cleanup - tier-0 TTL runtime pruning added for single-field non-partial TTL indexes - deterministic R3 ledger mismatches reduced for `_id` dollar subfields, `updateMany` replacement validation, and transactional `createIndexes` +- analytics-oriented update/aggregation parity expanded with `$min`/`$max`, expression-based pipeline updates, `$dateTrunc`, percentile/median and N accumulators, and a deterministic `$setWindowFields` subset ## Current Focus @@ -69,7 +70,7 @@ Primary measures: - close remaining aggregate-stage unsupported surface beyond the current `$merge` terminal subset - expand collation semantics beyond current subset (`locale`/`strength`/`caseLevel`) - expand TTL runtime behavior beyond the current single-field lazy-prune subset -- expand update/operator coverage beyond current `arrayFilters` subset (advanced positional/pipeline expressions) +- expand update/operator coverage beyond current `arrayFilters` and pipeline-expression subsets (advanced positional updates and remaining expressions) - expand supported transaction operations in unified suites while preserving deterministic behavior ## Out of Scope (Current Phase) diff --git a/docs/SUPPORT_MATRIX.md b/docs/SUPPORT_MATRIX.md index 882aaf3..c8ab49d 100644 --- a/docs/SUPPORT_MATRIX.md +++ b/docs/SUPPORT_MATRIX.md @@ -1,6 +1,6 @@ # Support Matrix -Status date: 2026-05-09 +Status date: 2026-07-21 This matrix is a versioned support boundary for integration-test usage. Source artifact: `build/reports/r2-compatibility/r2-support-manifest.json`. @@ -20,9 +20,9 @@ Source artifact: `build/reports/r2-compatibility/r2-support-manifest.json`. | `query.eq-ne-compare` | query | Supported | Core comparison operators | | `query.elemMatch-all-regex` | query | Supported | Array and regex operators | | `query.expr-subset` | query | Partial | Subset: eq/ne/gt/gte/lt/lte/and/or/not/literal/add | -| `aggregation.match-project-group` | aggregation | Supported | Tier-1 pipeline stages, including `$unwind.includeArrayIndex` | -| `aggregation.lookup-union-facet` | aggregation | Partial | Tier-2 subset without full expression parity (includes minimal `$graphLookup` option subset) | -| `aggregation.expression-operators` | aggregation | Partial | Limited expression coverage | +| `aggregation.match-project-group` | aggregation | Supported | Tier-1 pipeline stages, analytics accumulators, and `$unwind.includeArrayIndex` | +| `aggregation.lookup-union-facet` | aggregation | Partial | Tier-2 subset without full expression parity (includes minimal `$graphLookup` and `$setWindowFields` option subsets) | +| `aggregation.expression-operators` | aggregation | Partial | Conditional/comparison/type/null, min/max, arithmetic, mergeObjects, and UTC dateTrunc subsets | | `index.unique-sparse-partial` | index | Supported | Unique/sparse/partial | | `index.collation-metadata` | index | Supported | Collation metadata round-trip | | `index.collation-semantic` | index | Partial | Subset: locale/strength/caseLevel on query-sort-distinct and unique index checks | diff --git a/jongodb-spring-suite/build.gradle.kts b/jongodb-spring-suite/build.gradle.kts index 6b9fc75..f9955dd 100644 --- a/jongodb-spring-suite/build.gradle.kts +++ b/jongodb-spring-suite/build.gradle.kts @@ -25,6 +25,7 @@ dependencies { testImplementation(platform("org.junit:junit-bom:5.10.2")) testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") } tasks.withType().configureEach { diff --git a/jongodb-testkit/build.gradle.kts b/jongodb-testkit/build.gradle.kts index 080e822..9c85d7b 100644 --- a/jongodb-testkit/build.gradle.kts +++ b/jongodb-testkit/build.gradle.kts @@ -29,6 +29,7 @@ dependencies { testImplementation(platform("org.junit:junit-bom:5.10.2")) testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") } tasks.withType().configureEach { diff --git a/src/main/java/org/jongodb/command/CommandStore.java b/src/main/java/org/jongodb/command/CommandStore.java index 40ed122..1e479b2 100644 --- a/src/main/java/org/jongodb/command/CommandStore.java +++ b/src/main/java/org/jongodb/command/CommandStore.java @@ -1,6 +1,7 @@ package org.jongodb.command; import java.util.List; +import org.bson.BsonArray; import org.bson.BsonDocument; import org.bson.BsonValue; import org.jongodb.engine.CollationSupport; @@ -146,13 +147,43 @@ public IndexMetadata( } record UpdateRequest( - BsonDocument query, BsonDocument update, boolean multi, boolean upsert, List arrayFilters) { + BsonDocument query, + BsonDocument update, + BsonArray updatePipeline, + boolean multi, + boolean upsert, + List arrayFilters) { public UpdateRequest(final BsonDocument query, final BsonDocument update, final boolean multi, final boolean upsert) { - this(query, update, multi, upsert, List.of()); + this(query, update, null, multi, upsert, List.of()); + } + + public UpdateRequest( + final BsonDocument query, + final BsonDocument update, + final boolean multi, + final boolean upsert, + final List arrayFilters) { + this(query, update, null, multi, upsert, arrayFilters); + } + + public UpdateRequest( + final BsonDocument query, + final BsonArray updatePipeline, + final boolean multi, + final boolean upsert) { + this(query, null, updatePipeline, multi, upsert, List.of()); } public UpdateRequest { + if ((update == null) == (updatePipeline == null)) { + throw new IllegalArgumentException("exactly one of update or updatePipeline must be specified"); + } + update = update == null ? null : update.clone(); + updatePipeline = updatePipeline == null ? null : updatePipeline.clone(); arrayFilters = copyArrayFilters(arrayFilters); + if (updatePipeline != null && !arrayFilters.isEmpty()) { + throw new IllegalArgumentException("arrayFilters is not allowed with pipeline updates"); + } } private static List copyArrayFilters(final List source) { diff --git a/src/main/java/org/jongodb/command/EngineBackedCommandStore.java b/src/main/java/org/jongodb/command/EngineBackedCommandStore.java index 2a20a3a..74a4e9a 100644 --- a/src/main/java/org/jongodb/command/EngineBackedCommandStore.java +++ b/src/main/java/org/jongodb/command/EngineBackedCommandStore.java @@ -246,14 +246,24 @@ public UpdateResult update(final String database, final String collection, final for (int index = 0; index < updates.size(); index++) { final UpdateRequest updateRequest = updates.get(index); final Document query = toDocument(Objects.requireNonNull(updateRequest.query(), "query")); - final Document update = toDocument(Objects.requireNonNull(updateRequest.update(), "update")); final List arrayFilters = new ArrayList<>(updateRequest.arrayFilters().size()); for (final BsonDocument arrayFilter : updateRequest.arrayFilters()) { arrayFilters.add(toDocument(Objects.requireNonNull(arrayFilter, "arrayFilters entries must not be null"))); } - final UpdateManyResult result = collectionStore.update( - query, update, updateRequest.multi(), updateRequest.upsert(), List.copyOf(arrayFilters)); + final UpdateManyResult result; + if (updateRequest.updatePipeline() != null) { + final List pipeline = new ArrayList<>(updateRequest.updatePipeline().size()); + for (final BsonValue stageValue : updateRequest.updatePipeline()) { + pipeline.add(toDocument(stageValue.asDocument())); + } + result = collectionStore.updatePipeline( + query, List.copyOf(pipeline), updateRequest.multi(), updateRequest.upsert()); + } else { + final Document update = toDocument(Objects.requireNonNull(updateRequest.update(), "update")); + result = collectionStore.update( + query, update, updateRequest.multi(), updateRequest.upsert(), List.copyOf(arrayFilters)); + } matchedCount += toBoundedInt(result.matchedCount()); modifiedCount += toBoundedInt(result.modifiedCount()); if (result.upserted()) { diff --git a/src/main/java/org/jongodb/command/FindAndModifyCommandHandler.java b/src/main/java/org/jongodb/command/FindAndModifyCommandHandler.java index 10b1a40..b372b78 100644 --- a/src/main/java/org/jongodb/command/FindAndModifyCommandHandler.java +++ b/src/main/java/org/jongodb/command/FindAndModifyCommandHandler.java @@ -4,6 +4,7 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Set; +import org.bson.BsonArray; import org.bson.BsonBoolean; import org.bson.BsonDocument; import org.bson.BsonDouble; @@ -101,10 +102,20 @@ public BsonDocument handle(final BsonDocument command) { return CommandErrors.badValue("upsert is not allowed when remove=true"); } } - if (!remove && (updateValue == null || !updateValue.isDocument())) { - return CommandErrors.typeMismatch("update must be a document"); + if (!remove && (updateValue == null || (!updateValue.isDocument() && !updateValue.isArray()))) { + return CommandErrors.typeMismatch("update must be a document or array"); } final BsonDocument update = updateValue != null && updateValue.isDocument() ? updateValue.asDocument() : null; + final BsonArray updatePipeline; + if (updateValue != null && updateValue.isArray()) { + final UpdatePipelineSubset.ParseResult parsedPipeline = UpdatePipelineSubset.parse(updateValue.asArray()); + if (parsedPipeline.error() != null) { + return parsedPipeline.error(); + } + updatePipeline = parsedPipeline.updatePipeline(); + } else { + updatePipeline = null; + } final UpdateArrayFiltersSubset.ParseResult parsedArrayFilters = UpdateArrayFiltersSubset.parse(command.get("arrayFilters")); if (parsedArrayFilters.error() != null) { @@ -113,6 +124,9 @@ public BsonDocument handle(final BsonDocument command) { if (remove && !parsedArrayFilters.parsed().isEmpty()) { return CommandErrors.badValue("arrayFilters is not allowed when remove=true"); } + if (updatePipeline != null && !parsedArrayFilters.parsed().isEmpty()) { + return CommandErrors.badValue("arrayFilters is not allowed with pipeline updates"); + } final List matches; try { @@ -132,6 +146,7 @@ public BsonDocument handle(final BsonDocument command) { collection, query, update, + updatePipeline, selected, upsert, returnNew, @@ -167,6 +182,7 @@ private BsonDocument handleUpdate( final String collection, final BsonDocument query, final BsonDocument update, + final BsonArray updatePipeline, final BsonDocument selected, final boolean upsert, final boolean returnNew, @@ -178,7 +194,9 @@ private BsonDocument handleUpdate( final CommandStore.UpdateResult result = store.update( database, collection, - List.of(new CommandStore.UpdateRequest(oneFilter, update, false, false, arrayFilters))); + List.of(updatePipeline == null + ? new CommandStore.UpdateRequest(oneFilter, update, false, false, arrayFilters) + : new CommandStore.UpdateRequest(oneFilter, updatePipeline, false, false))); final BsonDocument selectedValue = returnNew ? firstMatch(database, collection, oneFilter, collation) : selected; final BsonDocument value = applyProjection(selectedValue, projectionSpec); return successResponse(result.matchedCount() > 0 ? 1 : 0, result.matchedCount() > 0, null, value); @@ -191,7 +209,9 @@ private BsonDocument handleUpdate( final CommandStore.UpdateResult result = store.update( database, collection, - List.of(new CommandStore.UpdateRequest(query, update, false, true, arrayFilters))); + List.of(updatePipeline == null + ? new CommandStore.UpdateRequest(query, update, false, true, arrayFilters) + : new CommandStore.UpdateRequest(query, updatePipeline, false, true))); final BsonValue upsertedId = result.upserted().isEmpty() ? null : result.upserted().get(0).id(); diff --git a/src/main/java/org/jongodb/command/FindOneAndUpdateCommandHandler.java b/src/main/java/org/jongodb/command/FindOneAndUpdateCommandHandler.java index 4ce6775..74bbc64 100644 --- a/src/main/java/org/jongodb/command/FindOneAndUpdateCommandHandler.java +++ b/src/main/java/org/jongodb/command/FindOneAndUpdateCommandHandler.java @@ -8,7 +8,8 @@ import org.jongodb.command.CommandCanonicalizer.ValidationException; public final class FindOneAndUpdateCommandHandler implements CommandHandler { - private static final Set SUPPORTED_OPERATORS = Set.of("$set", "$setOnInsert", "$inc", "$unset", "$addToSet"); + private static final Set SUPPORTED_OPERATORS = + Set.of("$set", "$setOnInsert", "$inc", "$min", "$max", "$unset", "$addToSet"); private final FindAndModifyCommandHandler findAndModifyCommandHandler; @@ -49,17 +50,20 @@ public BsonDocument handle(final BsonDocument command) { } final BsonValue updateValue = command.get("update"); - final BsonDocument update; + final BsonValue update; + final boolean pipelineUpdate; if (updateValue == null) { return CommandErrors.typeMismatch("update must be a document or array"); } else if (updateValue.isDocument()) { update = updateValue.asDocument(); + pipelineUpdate = false; } else if (updateValue.isArray()) { final UpdatePipelineSubset.ParseResult parsedPipeline = UpdatePipelineSubset.parse(updateValue.asArray()); if (parsedPipeline.error() != null) { return parsedPipeline.error(); } - update = parsedPipeline.updateDocument(); + update = parsedPipeline.updatePipeline(); + pipelineUpdate = true; } else { return CommandErrors.typeMismatch("update must be a document or array"); } @@ -85,10 +89,15 @@ public BsonDocument handle(final BsonDocument command) { if (parsedArrayFilters.error() != null) { return parsedArrayFilters.error(); } + if (pipelineUpdate && !parsedArrayFilters.parsed().isEmpty()) { + return CommandErrors.badValue("arrayFilters is not allowed with pipeline updates"); + } - optionError = validateOperatorUpdateDocument(update, parsedArrayFilters.parsed()); - if (optionError != null) { - return optionError; + if (!pipelineUpdate) { + optionError = validateOperatorUpdateDocument(update.asDocument(), parsedArrayFilters.parsed()); + if (optionError != null) { + return optionError; + } } final boolean returnNew; @@ -158,6 +167,16 @@ private static BsonDocument validateOperatorUpdateDocument( return CommandErrors.typeMismatch("$inc must be a document"); } + final BsonValue minValue = update.get("$min"); + if (minValue != null && !minValue.isDocument()) { + return CommandErrors.typeMismatch("$min must be a document"); + } + + final BsonValue maxValue = update.get("$max"); + if (maxValue != null && !maxValue.isDocument()) { + return CommandErrors.typeMismatch("$max must be a document"); + } + final BsonValue unsetValue = update.get("$unset"); if (unsetValue != null && !unsetValue.isDocument()) { return CommandErrors.typeMismatch("$unset must be a document"); diff --git a/src/main/java/org/jongodb/command/UpdateArrayFiltersSubset.java b/src/main/java/org/jongodb/command/UpdateArrayFiltersSubset.java index c009a2d..d246266 100644 --- a/src/main/java/org/jongodb/command/UpdateArrayFiltersSubset.java +++ b/src/main/java/org/jongodb/command/UpdateArrayFiltersSubset.java @@ -62,6 +62,7 @@ static BsonDocument validateUpdatePaths( Objects.requireNonNull(parsedArrayFilters, "parsedArrayFilters"); final Set usedIdentifiers = new LinkedHashSet<>(); + final List claimedPaths = new ArrayList<>(); for (final String operator : updateDocument.keySet()) { final BsonValue operatorDefinition = updateDocument.get(operator); if (operatorDefinition == null || !operatorDefinition.isDocument()) { @@ -77,6 +78,15 @@ static BsonDocument validateUpdatePaths( return CommandErrors.badValue( "positional and array filter updates are not supported for path '" + path + "'"); } + for (final String claimedPath : claimedPaths) { + if (path.equals(claimedPath) + || path.startsWith(claimedPath + ".") + || claimedPath.startsWith(path + ".")) { + return CommandErrors.badValue( + "updating the path '" + path + "' would create a conflict at '" + claimedPath + "'"); + } + } + claimedPaths.add(path); if (analysis.identifiers().isEmpty()) { continue; } diff --git a/src/main/java/org/jongodb/command/UpdateCommandHandler.java b/src/main/java/org/jongodb/command/UpdateCommandHandler.java index 0fa689e..6559520 100644 --- a/src/main/java/org/jongodb/command/UpdateCommandHandler.java +++ b/src/main/java/org/jongodb/command/UpdateCommandHandler.java @@ -12,7 +12,8 @@ import org.jongodb.engine.DuplicateKeyException; public final class UpdateCommandHandler implements CommandHandler { - private static final Set SUPPORTED_OPERATORS = Set.of("$set", "$setOnInsert", "$inc", "$unset", "$addToSet"); + private static final Set SUPPORTED_OPERATORS = + Set.of("$set", "$setOnInsert", "$inc", "$min", "$max", "$unset", "$addToSet"); private final CommandStore store; @@ -65,16 +66,19 @@ public BsonDocument handle(final BsonDocument command) { final BsonValue updateValue = updateSpec.get("u"); final BsonDocument updateDocument; + final BsonArray updatePipeline; if (updateValue == null) { return CommandErrors.typeMismatch("u must be a document or array"); } else if (updateValue.isDocument()) { updateDocument = updateValue.asDocument(); + updatePipeline = null; } else if (updateValue.isArray()) { final UpdatePipelineSubset.ParseResult parsedPipeline = UpdatePipelineSubset.parse(updateValue.asArray()); if (parsedPipeline.error() != null) { return parsedPipeline.error(); } - updateDocument = parsedPipeline.updateDocument(); + updateDocument = null; + updatePipeline = parsedPipeline.updatePipeline(); } else { return CommandErrors.typeMismatch("u must be a document or array"); } @@ -113,15 +117,22 @@ public BsonDocument handle(final BsonDocument command) { if (parsedArrayFilters.error() != null) { return parsedArrayFilters.error(); } + if (updatePipeline != null && !parsedArrayFilters.parsed().isEmpty()) { + return CommandErrors.badValue("arrayFilters is not allowed with pipeline updates"); + } - final BsonDocument updateValidationError = - validateUpdateDocument(updateDocument, multi, parsedArrayFilters.parsed()); - if (updateValidationError != null) { - return updateValidationError; + if (updateDocument != null) { + final BsonDocument updateValidationError = + validateUpdateDocument(updateDocument, multi, parsedArrayFilters.parsed()); + if (updateValidationError != null) { + return updateValidationError; + } } - updates.add(new CommandStore.UpdateRequest( - query, updateDocument, multi, upsert, parsedArrayFilters.parsed().filters())); + updates.add(updatePipeline == null + ? new CommandStore.UpdateRequest( + query, updateDocument, multi, upsert, parsedArrayFilters.parsed().filters()) + : new CommandStore.UpdateRequest(query, updatePipeline, multi, upsert)); } final CommandStore.UpdateResult result; @@ -197,6 +208,16 @@ private static BsonDocument validateUpdateDocument( return CommandErrors.typeMismatch("$inc must be a document"); } + final BsonValue minValue = updateDocument.get("$min"); + if (minValue != null && !minValue.isDocument()) { + return CommandErrors.typeMismatch("$min must be a document"); + } + + final BsonValue maxValue = updateDocument.get("$max"); + if (maxValue != null && !maxValue.isDocument()) { + return CommandErrors.typeMismatch("$max must be a document"); + } + final BsonValue unsetValue = updateDocument.get("$unset"); if (unsetValue != null && !unsetValue.isDocument()) { return CommandErrors.typeMismatch("$unset must be a document"); diff --git a/src/main/java/org/jongodb/command/UpdatePipelineSubset.java b/src/main/java/org/jongodb/command/UpdatePipelineSubset.java index a624304..65635d2 100644 --- a/src/main/java/org/jongodb/command/UpdatePipelineSubset.java +++ b/src/main/java/org/jongodb/command/UpdatePipelineSubset.java @@ -1,16 +1,14 @@ package org.jongodb.command; -import java.util.LinkedHashMap; -import java.util.LinkedHashSet; -import java.util.Map; +import java.util.Set; import org.bson.BsonArray; import org.bson.BsonDocument; -import org.bson.BsonInt32; import org.bson.BsonValue; +/** Validation for the stages MongoDB permits in aggregation-pipeline updates. */ final class UpdatePipelineSubset { - private static final String STAGE_SET = "$set"; - private static final String STAGE_UNSET = "$unset"; + private static final Set SUPPORTED_STAGES = Set.of( + "$addFields", "$set", "$project", "$unset", "$replaceRoot", "$replaceWith"); private UpdatePipelineSubset() {} @@ -19,166 +17,45 @@ static ParseResult parse(final BsonArray pipeline) { return ParseResult.error(CommandErrors.badValue("update pipeline must not be empty")); } - final Map setOperations = new LinkedHashMap<>(); - final LinkedHashSet unsetOperations = new LinkedHashSet<>(); - + final BsonArray copied = new BsonArray(); for (final BsonValue stageValue : pipeline) { if (!stageValue.isDocument()) { return ParseResult.error(CommandErrors.typeMismatch("update pipeline stages must be documents")); } - final BsonDocument stage = stageValue.asDocument(); if (stage.size() != 1) { - return ParseResult.error(CommandErrors.badValue("update pipeline stage must contain exactly one operator")); + return ParseResult.error( + CommandErrors.badValue("update pipeline stage must contain exactly one operator")); } - final String stageName = stage.getFirstKey(); - final BsonValue stageArgument = stage.get(stageName); - if (STAGE_SET.equals(stageName)) { - final BsonDocument error = applySetStage(stageArgument, setOperations, unsetOperations); - if (error != null) { - return ParseResult.error(error); - } - continue; - } - if (STAGE_UNSET.equals(stageName)) { - final BsonDocument error = applyUnsetStage(stageArgument, setOperations, unsetOperations); - if (error != null) { - return ParseResult.error(error); - } - continue; - } - return ParseResult.error(CommandErrors.badValue("unsupported update pipeline stage: " + stageName)); - } - - final BsonDocument normalized = new BsonDocument(); - if (!setOperations.isEmpty()) { - final BsonDocument setDocument = new BsonDocument(); - for (final Map.Entry entry : setOperations.entrySet()) { - setDocument.append(entry.getKey(), entry.getValue()); - } - normalized.append(STAGE_SET, setDocument); - } - if (!unsetOperations.isEmpty()) { - final BsonDocument unsetDocument = new BsonDocument(); - for (final String path : unsetOperations) { - unsetDocument.append(path, new BsonInt32(1)); - } - normalized.append(STAGE_UNSET, unsetDocument); - } - if (normalized.isEmpty()) { - return ParseResult.error(CommandErrors.badValue("update pipeline must include at least one field operation")); - } - return ParseResult.success(normalized); - } - - private static BsonDocument applySetStage( - final BsonValue stageArgument, - final Map setOperations, - final LinkedHashSet unsetOperations) { - if (stageArgument == null || !stageArgument.isDocument()) { - return CommandErrors.typeMismatch("$set stage must be a document"); - } - - for (final Map.Entry entry : stageArgument.asDocument().entrySet()) { - final String path = entry.getKey(); - if (containsUnsupportedExpression(entry.getValue())) { - return CommandErrors.badValue("update pipeline expressions are not supported for path '" + path + "'"); - } - unsetOperations.remove(path); - setOperations.put(path, entry.getValue()); - } - return null; - } - - private static BsonDocument applyUnsetStage( - final BsonValue stageArgument, - final Map setOperations, - final LinkedHashSet unsetOperations) { - if (stageArgument == null) { - return CommandErrors.typeMismatch("$unset stage must be a string, array, or document"); - } - - if (stageArgument.isString()) { - markUnset(stageArgument.asString().getValue(), setOperations, unsetOperations); - return null; - } - if (stageArgument.isArray()) { - for (final BsonValue value : stageArgument.asArray()) { - if (!value.isString()) { - return CommandErrors.typeMismatch("$unset stage array entries must be strings"); - } - markUnset(value.asString().getValue(), setOperations, unsetOperations); - } - return null; - } - if (!stageArgument.isDocument()) { - return CommandErrors.typeMismatch("$unset stage must be a string, array, or document"); - } - - for (final String path : stageArgument.asDocument().keySet()) { - markUnset(path, setOperations, unsetOperations); - } - return null; - } - - private static void markUnset( - final String path, - final Map setOperations, - final LinkedHashSet unsetOperations) { - setOperations.remove(path); - unsetOperations.add(path); - } - - private static boolean containsUnsupportedExpression(final BsonValue value) { - if (value == null) { - return false; - } - if (value.isString()) { - final String raw = value.asString().getValue(); - return raw != null && raw.startsWith("$"); - } - if (value.isDocument()) { - for (final Map.Entry entry : value.asDocument().entrySet()) { - if (entry.getKey().startsWith("$")) { - return true; - } - if (containsUnsupportedExpression(entry.getValue())) { - return true; - } - } - return false; - } - if (value.isArray()) { - for (final BsonValue item : value.asArray()) { - if (containsUnsupportedExpression(item)) { - return true; - } + if (!SUPPORTED_STAGES.contains(stageName)) { + return ParseResult.error( + CommandErrors.badValue("unsupported update pipeline stage: " + stageName)); } - return false; + copied.add(stage.clone()); } - return false; + return ParseResult.success(copied); } static final class ParseResult { - private final BsonDocument updateDocument; + private final BsonArray updatePipeline; private final BsonDocument error; - private ParseResult(final BsonDocument updateDocument, final BsonDocument error) { - this.updateDocument = updateDocument; + private ParseResult(final BsonArray updatePipeline, final BsonDocument error) { + this.updatePipeline = updatePipeline; this.error = error; } - static ParseResult success(final BsonDocument updateDocument) { - return new ParseResult(updateDocument, null); + static ParseResult success(final BsonArray updatePipeline) { + return new ParseResult(updatePipeline, null); } static ParseResult error(final BsonDocument error) { return new ParseResult(null, error); } - BsonDocument updateDocument() { - return updateDocument; + BsonArray updatePipeline() { + return updatePipeline == null ? null : updatePipeline.clone(); } BsonDocument error() { diff --git a/src/main/java/org/jongodb/engine/AggregationExpressions.java b/src/main/java/org/jongodb/engine/AggregationExpressions.java new file mode 100644 index 0000000..6317224 --- /dev/null +++ b/src/main/java/org/jongodb/engine/AggregationExpressions.java @@ -0,0 +1,603 @@ +package org.jongodb.engine; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.time.DayOfWeek; +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.time.temporal.ChronoUnit; +import java.time.temporal.TemporalAdjusters; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; +import org.bson.BsonRegularExpression; +import org.bson.BsonTimestamp; +import org.bson.Document; +import org.bson.types.Binary; +import org.bson.types.Decimal128; +import org.bson.types.ObjectId; + +/** Expression evaluator used by aggregation stages and aggregation-pipeline updates. */ +final class AggregationExpressions { + private static final Object MISSING = new Object(); + private static final ZonedDateTime REFERENCE = + ZonedDateTime.of(2000, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); + + private AggregationExpressions() {} + + static Object evaluate(final Document source, final Object expression) { + if (expression instanceof String pathExpression && pathExpression.startsWith("$")) { + if ("$$ROOT".equals(pathExpression) || "$$CURRENT".equals(pathExpression)) { + return DocumentCopies.copy(source); + } + if (pathExpression.startsWith("$$")) { + throw new UnsupportedFeatureException( + "aggregation.expression.variable", + "unsupported aggregation variable: " + pathExpression); + } + return resolvePath(source, pathExpression.substring(1)); + } + if (expression instanceof Map rawMap) { + final Map map = stringMap(rawMap); + if (map.size() == 1) { + final Map.Entry entry = map.entrySet().iterator().next(); + if (entry.getKey().startsWith("$")) { + return evaluateOperator(source, entry.getKey(), entry.getValue()); + } + } + final Document evaluated = new Document(); + for (final Map.Entry entry : map.entrySet()) { + final Object value = evaluate(source, entry.getValue()); + if (!isMissing(value)) { + evaluated.put(entry.getKey(), DocumentCopies.copyAny(value)); + } + } + return evaluated; + } + if (expression instanceof List list) { + final List evaluated = new ArrayList<>(list.size()); + for (final Object item : list) { + final Object value = evaluate(source, item); + evaluated.add(isMissing(value) ? null : DocumentCopies.copyAny(value)); + } + return evaluated; + } + return DocumentCopies.copyAny(expression); + } + + static boolean isMissing(final Object value) { + return value == MISSING; + } + + static Object nullIfMissing(final Object value) { + return isMissing(value) ? null : value; + } + + static boolean truthy(final Object value) { + if (value == null || isMissing(value)) { + return false; + } + if (value instanceof Boolean booleanValue) { + return booleanValue; + } + if (value instanceof Number numberValue) { + return numberValue.doubleValue() != 0d && !Double.isNaN(numberValue.doubleValue()); + } + return true; + } + + private static Object evaluateOperator( + final Document source, final String operator, final Object operand) { + return switch (operator) { + case "$literal" -> DocumentCopies.copyAny(operand); + case "$cond" -> evaluateCond(source, operand); + case "$ifNull" -> evaluateIfNull(source, operand); + case "$type" -> typeName(evaluate(source, operand)); + case "$eq" -> compare(source, operand, Comparison.EQ); + case "$ne" -> compare(source, operand, Comparison.NE); + case "$gt" -> compare(source, operand, Comparison.GT); + case "$gte" -> compare(source, operand, Comparison.GTE); + case "$lt" -> compare(source, operand, Comparison.LT); + case "$lte" -> compare(source, operand, Comparison.LTE); + case "$cmp" -> compare(source, operand, Comparison.CMP); + case "$and" -> evaluateAnd(source, operand); + case "$or" -> evaluateOr(source, operand); + case "$not" -> evaluateNot(source, operand); + case "$max" -> evaluateMinMax(source, operand, true); + case "$min" -> evaluateMinMax(source, operand, false); + case "$add" -> evaluateArithmetic(source, operand, Arithmetic.ADD); + case "$subtract" -> evaluateArithmetic(source, operand, Arithmetic.SUBTRACT); + case "$multiply" -> evaluateArithmetic(source, operand, Arithmetic.MULTIPLY); + case "$divide" -> evaluateArithmetic(source, operand, Arithmetic.DIVIDE); + case "$mergeObjects" -> evaluateMergeObjects(source, operand); + case "$dateTrunc" -> evaluateDateTrunc(source, operand); + default -> throw new UnsupportedFeatureException( + "aggregation.expression." + operator, + "unsupported aggregation expression: " + operator); + }; + } + + private static Object evaluateCond(final Document source, final Object operand) { + final Object condition; + final Object thenExpression; + final Object elseExpression; + if (operand instanceof List list) { + if (list.size() != 3) { + throw new IllegalArgumentException("$cond array form requires exactly three arguments"); + } + condition = list.get(0); + thenExpression = list.get(1); + elseExpression = list.get(2); + } else if (operand instanceof Map map) { + final Map definition = stringMap(map); + if (!definition.keySet().equals(java.util.Set.of("if", "then", "else"))) { + throw new IllegalArgumentException("$cond object form requires if, then, and else"); + } + condition = definition.get("if"); + thenExpression = definition.get("then"); + elseExpression = definition.get("else"); + } else { + throw new IllegalArgumentException("$cond requires an array or document argument"); + } + return evaluate(source, truthy(evaluate(source, condition)) ? thenExpression : elseExpression); + } + + private static Object evaluateIfNull(final Document source, final Object operand) { + final List arguments = requireArguments(operand, "$ifNull", 2, null); + for (final Object argument : arguments) { + final Object value = evaluate(source, argument); + if (value != null && !isMissing(value)) { + return value; + } + } + return null; + } + + private static Object compare( + final Document source, final Object operand, final Comparison comparison) { + final List arguments = requireArguments(operand, comparison.operatorName(), 2, 2); + final Object left = nullIfMissing(evaluate(source, arguments.get(0))); + final Object right = nullIfMissing(evaluate(source, arguments.get(1))); + final int compared = MongoValueComparator.compare(left, right); + return switch (comparison) { + case EQ -> compared == 0; + case NE -> compared != 0; + case GT -> compared > 0; + case GTE -> compared >= 0; + case LT -> compared < 0; + case LTE -> compared <= 0; + case CMP -> Integer.signum(compared); + }; + } + + private static boolean evaluateAnd(final Document source, final Object operand) { + for (final Object argument : requireArguments(operand, "$and", 0, null)) { + if (!truthy(evaluate(source, argument))) { + return false; + } + } + return true; + } + + private static boolean evaluateOr(final Document source, final Object operand) { + for (final Object argument : requireArguments(operand, "$or", 0, null)) { + if (truthy(evaluate(source, argument))) { + return true; + } + } + return false; + } + + private static boolean evaluateNot(final Document source, final Object operand) { + final List arguments = requireArguments(operand, "$not", 1, 1); + return !truthy(evaluate(source, arguments.get(0))); + } + + private static Object evaluateMinMax( + final Document source, final Object operand, final boolean maximum) { + final List values = new ArrayList<>(); + if (operand instanceof List arguments) { + for (final Object argument : arguments) { + collectMinMaxValue(values, evaluate(source, argument), false); + } + } else { + collectMinMaxValue(values, evaluate(source, operand), true); + } + + Object selected = MISSING; + for (final Object value : values) { + if (value == null || isMissing(value)) { + continue; + } + if (isMissing(selected)) { + selected = value; + continue; + } + final int compared = MongoValueComparator.compare(value, selected); + if ((maximum && compared > 0) || (!maximum && compared < 0)) { + selected = value; + } + } + return isMissing(selected) ? null : DocumentCopies.copyAny(selected); + } + + private static void collectMinMaxValue( + final List values, final Object value, final boolean traverseArray) { + if (traverseArray && value instanceof List list) { + values.addAll(list); + return; + } + values.add(value); + } + + private static Object evaluateArithmetic( + final Document source, final Object operand, final Arithmetic arithmetic) { + final int minimum = arithmetic == Arithmetic.ADD || arithmetic == Arithmetic.MULTIPLY ? 1 : 2; + final Integer exact = arithmetic == Arithmetic.SUBTRACT || arithmetic == Arithmetic.DIVIDE ? 2 : null; + final List arguments = requireArguments(operand, arithmetic.operatorName(), minimum, exact); + final List numbers = new ArrayList<>(arguments.size()); + boolean floating = false; + for (final Object argument : arguments) { + final Object value = evaluate(source, argument); + if (value == null || isMissing(value)) { + return null; + } + if (!(value instanceof Number number)) { + throw new IllegalArgumentException(arithmetic.operatorName() + " accepts numeric operands only"); + } + floating |= number instanceof Float || number instanceof Double || number instanceof BigDecimal; + numbers.add(number); + } + + BigDecimal result = toBigDecimal(numbers.get(0)); + if (arithmetic == Arithmetic.ADD) { + result = BigDecimal.ZERO; + } else if (arithmetic == Arithmetic.MULTIPLY) { + result = BigDecimal.ONE; + } + for (int index = arithmetic == Arithmetic.SUBTRACT || arithmetic == Arithmetic.DIVIDE ? 1 : 0; + index < numbers.size(); + index++) { + final BigDecimal value = toBigDecimal(numbers.get(index)); + result = switch (arithmetic) { + case ADD -> result.add(value); + case SUBTRACT -> result.subtract(value); + case MULTIPLY -> result.multiply(value); + case DIVIDE -> { + if (value.compareTo(BigDecimal.ZERO) == 0) { + throw new IllegalArgumentException("$divide cannot divide by zero"); + } + yield result.divide(value, java.math.MathContext.DECIMAL128); + } + }; + } + if (arithmetic == Arithmetic.DIVIDE || floating) { + return result.doubleValue(); + } + try { + return result.longValueExact(); + } catch (final ArithmeticException ignored) { + return result; + } + } + + private static Object evaluateMergeObjects(final Document source, final Object operand) { + final List expressions = operand instanceof List list ? list : List.of(operand); + final Document merged = new Document(); + for (final Object expression : expressions) { + final Object value = evaluate(source, expression); + if (value == null || isMissing(value)) { + continue; + } + if (!(value instanceof Map map)) { + throw new IllegalArgumentException("$mergeObjects operands must evaluate to documents or null"); + } + for (final Map.Entry entry : map.entrySet()) { + if (!(entry.getKey() instanceof String key)) { + throw new IllegalArgumentException("$mergeObjects document keys must be strings"); + } + merged.put(key, DocumentCopies.copyAny(entry.getValue())); + } + } + return merged; + } + + private static Object evaluateDateTrunc(final Document source, final Object operand) { + if (!(operand instanceof Map rawDefinition)) { + throw new IllegalArgumentException("$dateTrunc requires a document argument"); + } + final Map definition = stringMap(rawDefinition); + for (final String key : definition.keySet()) { + if (!java.util.Set.of("date", "unit", "binSize", "timezone", "startOfWeek").contains(key)) { + throw new IllegalArgumentException("unsupported $dateTrunc option: " + key); + } + } + if (!definition.containsKey("date") || !definition.containsKey("unit")) { + throw new IllegalArgumentException("$dateTrunc requires date and unit"); + } + + final Object rawDate = evaluate(source, definition.get("date")); + if (rawDate == null || isMissing(rawDate)) { + return null; + } + if (!(rawDate instanceof Date) + && !(rawDate instanceof Instant) + && !(rawDate instanceof BsonTimestamp) + && !(rawDate instanceof ObjectId)) { + throw new IllegalArgumentException("$dateTrunc.date must evaluate to a date"); + } + final Instant instant; + if (rawDate instanceof Date date) { + instant = date.toInstant(); + } else if (rawDate instanceof Instant instantValue) { + instant = instantValue; + } else if (rawDate instanceof BsonTimestamp timestamp) { + instant = Instant.ofEpochSecond(Integer.toUnsignedLong(timestamp.getTime())); + } else { + instant = ((ObjectId) rawDate).getDate().toInstant(); + } + + final Object unitValue = evaluate(source, definition.get("unit")); + if (unitValue == null || isMissing(unitValue)) { + return null; + } + if (!(unitValue instanceof String unit)) { + throw new IllegalArgumentException("$dateTrunc.unit must evaluate to a string"); + } + final Object binSizeValue = definition.containsKey("binSize") + ? evaluate(source, definition.get("binSize")) + : 1; + if (binSizeValue == null || isMissing(binSizeValue)) { + return null; + } + final int binSize = readPositiveInt(binSizeValue, "$dateTrunc.binSize"); + final Object timezoneValue = definition.containsKey("timezone") + ? evaluate(source, definition.get("timezone")) + : "UTC"; + if (timezoneValue == null || isMissing(timezoneValue)) { + return null; + } + if (!(timezoneValue instanceof String timezone)) { + throw new IllegalArgumentException("$dateTrunc.timezone must evaluate to a string"); + } + if (!isUtc(timezone)) { + throw new UnsupportedFeatureException( + "aggregation.expression.$dateTrunc.timezone", + "$dateTrunc currently supports UTC timezone only: " + timezone); + } + final DayOfWeek startOfWeek; + if ("week".equalsIgnoreCase(unit)) { + final Object startOfWeekValue = definition.containsKey("startOfWeek") + ? evaluate(source, definition.get("startOfWeek")) + : "sunday"; + if (startOfWeekValue == null || isMissing(startOfWeekValue)) { + return null; + } + startOfWeek = parseDayOfWeek(startOfWeekValue); + } else { + startOfWeek = DayOfWeek.SUNDAY; + } + + return Date.from(truncateUtc(instant, unit.toLowerCase(java.util.Locale.ROOT), binSize, startOfWeek)); + } + + private static Instant truncateUtc( + final Instant instant, + final String unit, + final int binSize, + final DayOfWeek startOfWeek) { + final ZonedDateTime value = instant.atZone(ZoneOffset.UTC); + final ZonedDateTime truncated; + switch (unit) { + case "second", "minute", "hour", "day" -> { + final long unitMillis = switch (unit) { + case "second" -> 1_000L; + case "minute" -> 60_000L; + case "hour" -> 3_600_000L; + default -> 86_400_000L; + }; + final long widthMillis = Math.multiplyExact(unitMillis, binSize); + final long elapsedMillis = value.toInstant().toEpochMilli() - REFERENCE.toInstant().toEpochMilli(); + truncated = REFERENCE.plus( + Math.floorDiv(elapsedMillis, widthMillis) * widthMillis, + ChronoUnit.MILLIS); + } + case "week" -> { + final ZonedDateTime weekReference = REFERENCE.with(TemporalAdjusters.nextOrSame(startOfWeek)); + final long widthMillis = Math.multiplyExact(604_800_000L, binSize); + final long elapsedMillis = value.toInstant().toEpochMilli() + - weekReference.toInstant().toEpochMilli(); + truncated = weekReference.plus( + Math.floorDiv(elapsedMillis, widthMillis) * widthMillis, + ChronoUnit.MILLIS); + } + case "month", "quarter", "year" -> { + final int unitMonths = "month".equals(unit) ? 1 : "quarter".equals(unit) ? 3 : 12; + final long elapsedMonths = ChronoUnit.MONTHS.between( + REFERENCE.toLocalDate().withDayOfMonth(1), value.toLocalDate().withDayOfMonth(1)); + final long width = (long) binSize * unitMonths; + truncated = REFERENCE.plusMonths(Math.floorDiv(elapsedMonths, width) * width); + } + default -> throw new IllegalArgumentException("unsupported $dateTrunc unit: " + unit); + } + return truncated.toInstant(); + } + + private static DayOfWeek parseDayOfWeek(final Object value) { + if (!(value instanceof String text)) { + throw new IllegalArgumentException("$dateTrunc.startOfWeek must evaluate to a string"); + } + return switch (text.toLowerCase(java.util.Locale.ROOT)) { + case "monday", "mon" -> DayOfWeek.MONDAY; + case "tuesday", "tue" -> DayOfWeek.TUESDAY; + case "wednesday", "wed" -> DayOfWeek.WEDNESDAY; + case "thursday", "thu" -> DayOfWeek.THURSDAY; + case "friday", "fri" -> DayOfWeek.FRIDAY; + case "saturday", "sat" -> DayOfWeek.SATURDAY; + case "sunday", "sun" -> DayOfWeek.SUNDAY; + default -> throw new IllegalArgumentException("unsupported $dateTrunc.startOfWeek: " + text); + }; + } + + private static boolean isUtc(final String timezone) { + return "UTC".equalsIgnoreCase(timezone) + || "GMT".equalsIgnoreCase(timezone) + || "Etc/UTC".equalsIgnoreCase(timezone) + || "Z".equalsIgnoreCase(timezone) + || "+00:00".equals(timezone) + || "-00:00".equals(timezone); + } + + private static int readPositiveInt(final Object value, final String name) { + if (!(value instanceof Number number) + || !Double.isFinite(number.doubleValue()) + || Math.rint(number.doubleValue()) != number.doubleValue() + || number.doubleValue() < 1 + || number.doubleValue() > Integer.MAX_VALUE) { + throw new IllegalArgumentException(name + " must be a positive integer"); + } + return number.intValue(); + } + + private static Object resolvePath(final Object source, final String path) { + if (path == null || path.isEmpty()) { + return MISSING; + } + Object current = source; + for (final String segment : path.split("\\.")) { + if (!(current instanceof Map map) || !map.containsKey(segment)) { + return MISSING; + } + current = map.get(segment); + } + return DocumentCopies.copyAny(current); + } + + private static String typeName(final Object value) { + if (isMissing(value)) { + return "missing"; + } + if (value == null) { + return "null"; + } + if (value instanceof Double || value instanceof Float) { + return "double"; + } + if (value instanceof Integer || value instanceof Short || value instanceof Byte) { + return "int"; + } + if (value instanceof Long || value instanceof BigInteger) { + return "long"; + } + if (value instanceof Decimal128 || value instanceof BigDecimal) { + return "decimal"; + } + if (value instanceof String || value instanceof Character) { + return "string"; + } + if (value instanceof Map) { + return "object"; + } + if (value instanceof List || value.getClass().isArray()) { + return "array"; + } + if (value instanceof Boolean) { + return "bool"; + } + if (value instanceof Date || value instanceof Instant) { + return "date"; + } + if (value instanceof ObjectId) { + return "objectId"; + } + if (value instanceof Binary || value instanceof byte[]) { + return "binData"; + } + if (value instanceof Pattern || value instanceof BsonRegularExpression) { + return "regex"; + } + return value.getClass().getSimpleName().toLowerCase(java.util.Locale.ROOT); + } + + private static List requireArguments( + final Object operand, + final String operator, + final int minimum, + final Integer exact) { + if (!(operand instanceof List list)) { + throw new IllegalArgumentException(operator + " requires an array argument"); + } + if (exact != null && list.size() != exact) { + throw new IllegalArgumentException(operator + " requires exactly " + exact + " arguments"); + } + if (list.size() < minimum) { + throw new IllegalArgumentException(operator + " requires at least " + minimum + " arguments"); + } + return list; + } + + private static Map stringMap(final Map rawMap) { + final Map output = new LinkedHashMap<>(); + for (final Map.Entry entry : rawMap.entrySet()) { + if (!(entry.getKey() instanceof String key)) { + throw new IllegalArgumentException("aggregation expression document keys must be strings"); + } + output.put(key, entry.getValue()); + } + return output; + } + + private static BigDecimal toBigDecimal(final Number value) { + if (value instanceof BigDecimal bigDecimal) { + return bigDecimal; + } + if (value instanceof BigInteger bigInteger) { + return new BigDecimal(bigInteger); + } + if (value instanceof Decimal128 decimal128) { + return decimal128.bigDecimalValue(); + } + return new BigDecimal(value.toString()); + } + + private enum Comparison { + EQ("$eq"), + NE("$ne"), + GT("$gt"), + GTE("$gte"), + LT("$lt"), + LTE("$lte"), + CMP("$cmp"); + + private final String operatorName; + + Comparison(final String operatorName) { + this.operatorName = operatorName; + } + + private String operatorName() { + return operatorName; + } + } + + private enum Arithmetic { + ADD("$add"), + SUBTRACT("$subtract"), + MULTIPLY("$multiply"), + DIVIDE("$divide"); + + private final String operatorName; + + Arithmetic(final String operatorName) { + this.operatorName = operatorName; + } + + private String operatorName() { + return operatorName; + } + } +} diff --git a/src/main/java/org/jongodb/engine/AggregationPipeline.java b/src/main/java/org/jongodb/engine/AggregationPipeline.java index 985ea82..d42041d 100644 --- a/src/main/java/org/jongodb/engine/AggregationPipeline.java +++ b/src/main/java/org/jongodb/engine/AggregationPipeline.java @@ -2,7 +2,6 @@ import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.Deque; import java.util.LinkedHashMap; @@ -107,6 +106,7 @@ public static List execute( case "$lookup" -> applyLookup(working, stageDefinition, collectionResolver, collation); case "$graphLookup" -> applyGraphLookup(working, stageDefinition, collectionResolver); case "$unionWith" -> applyUnionWith(working, stageDefinition, collectionResolver, collation); + case "$setWindowFields" -> applySetWindowFields(working, stageDefinition, collation); default -> throw new UnsupportedFeatureException( "aggregation.stage." + stageName, "unsupported aggregation stage: " + stageName); @@ -197,7 +197,10 @@ private static Document applyInclusionProjection(final Document source, final Do continue; } - output.put(field, evaluateExpression(source, entry.getValue())); + final Object evaluated = evaluateExpression(source, entry.getValue()); + if (!AggregationExpressions.isMissing(evaluated)) { + output.put(field, evaluated); + } } return output; } @@ -263,53 +266,22 @@ private static List applyGroup(final List input, final Objec final Object idExpression = groupDefinition.get("_id"); final List accumulators = parseAccumulators(groupDefinition); - final Map grouped = new LinkedHashMap<>(); + final Map grouped = new LinkedHashMap<>(); for (final Document source : input) { final Object id = evaluateGroupId(source, idExpression); final GroupKey groupKey = new GroupKey(id); - Document aggregate = grouped.get(groupKey); - if (aggregate == null) { - aggregate = new Document("_id", DocumentCopies.copyAny(id)); - for (final GroupAccumulator accumulator : accumulators) { - if (accumulator.operator() == GroupAccumulatorOperator.SUM) { - aggregate.put(accumulator.outputField(), 0L); - continue; - } - if (accumulator.operator() == GroupAccumulatorOperator.ADD_TO_SET) { - aggregate.put(accumulator.outputField(), new ArrayList<>()); - } - } - grouped.put(groupKey, aggregate); - } + grouped.computeIfAbsent(groupKey, ignored -> new GroupBucket(id)).sources().add(source); + } + final List output = new ArrayList<>(grouped.size()); + for (final GroupBucket bucket : grouped.values()) { + final Document aggregate = new Document("_id", DocumentCopies.copyAny(bucket.id())); for (final GroupAccumulator accumulator : accumulators) { - if (accumulator.operator() == GroupAccumulatorOperator.SUM) { - final Number increment = accumulator.sumOperand(source); - aggregate.put( - accumulator.outputField(), - addNumbers(aggregate.get(accumulator.outputField()), increment)); - continue; - } - - if (accumulator.operator() == GroupAccumulatorOperator.FIRST - && !aggregate.containsKey(accumulator.outputField())) { - aggregate.put(accumulator.outputField(), accumulator.firstOperand(source)); - continue; - } - - if (accumulator.operator() == GroupAccumulatorOperator.ADD_TO_SET) { - @SuppressWarnings("unchecked") - final List values = (List) aggregate.computeIfAbsent( - accumulator.outputField(), ignored -> new ArrayList<>()); - final Object candidate = accumulator.addToSetOperand(source); - if (!containsByMongoEquality(values, candidate)) { - values.add(candidate); - } - } + aggregate.put(accumulator.outputField(), evaluateAccumulator(accumulator, bucket.sources())); } + output.add(aggregate); } - - return List.copyOf(grouped.values()); + return List.copyOf(output); } private static List parseAccumulators(final Document groupDefinition) { @@ -328,8 +300,22 @@ private static List parseAccumulators(final Document groupDefi final String accumulatorName = accumulatorDefinition.keySet().iterator().next(); final GroupAccumulatorOperator operator = switch (accumulatorName) { case "$sum" -> GroupAccumulatorOperator.SUM; + case "$avg" -> GroupAccumulatorOperator.AVG; + case "$min" -> GroupAccumulatorOperator.MIN; + case "$max" -> GroupAccumulatorOperator.MAX; case "$first" -> GroupAccumulatorOperator.FIRST; + case "$last" -> GroupAccumulatorOperator.LAST; + case "$push" -> GroupAccumulatorOperator.PUSH; case "$addToSet" -> GroupAccumulatorOperator.ADD_TO_SET; + case "$mergeObjects" -> GroupAccumulatorOperator.MERGE_OBJECTS; + case "$percentile" -> GroupAccumulatorOperator.PERCENTILE; + case "$median" -> GroupAccumulatorOperator.MEDIAN; + case "$firstN" -> GroupAccumulatorOperator.FIRST_N; + case "$lastN" -> GroupAccumulatorOperator.LAST_N; + case "$minN" -> GroupAccumulatorOperator.MIN_N; + case "$maxN" -> GroupAccumulatorOperator.MAX_N; + case "$topN" -> GroupAccumulatorOperator.TOP_N; + case "$bottomN" -> GroupAccumulatorOperator.BOTTOM_N; default -> throw new UnsupportedFeatureException( "aggregation.group.accumulator." + accumulatorName, "unsupported $group accumulator: " + accumulatorName); @@ -340,66 +326,319 @@ private static List parseAccumulators(final Document groupDefi } private static Object evaluateGroupExpression(final Document source, final Object expression) { - if (expression instanceof String pathExpression && pathExpression.startsWith("$")) { - if ("$$ROOT".equals(pathExpression) || "$$CURRENT".equals(pathExpression)) { - return DocumentCopies.copy(source); + return AggregationExpressions.nullIfMissing(evaluateExpression(source, expression)); + } + + private enum GroupAccumulatorOperator { + SUM, + AVG, + MIN, + MAX, + FIRST, + LAST, + PUSH, + ADD_TO_SET, + MERGE_OBJECTS, + PERCENTILE, + MEDIAN, + FIRST_N, + LAST_N, + MIN_N, + MAX_N, + TOP_N, + BOTTOM_N + } + + private record GroupAccumulator(String outputField, GroupAccumulatorOperator operator, Object expression) {} + + private static Object evaluateGroupId(final Document source, final Object expression) { + return AggregationExpressions.nullIfMissing(evaluateExpression(source, expression)); + } + + private static Number addNumbers(final Object currentValue, final Number increment) { + if (!(currentValue instanceof Number currentNumber)) { + return normalizeNumber(increment.doubleValue()); + } + return normalizeNumber(currentNumber.doubleValue() + increment.doubleValue()); + } + + private static boolean containsByMongoEquality(final List values, final Object candidate) { + for (final Object value : values) { + if (MongoValueComparator.equals(value, candidate)) { + return true; } - final PathValue pathValue = resolvePath(source, pathExpression.substring(1)); - return pathValue.present() ? DocumentCopies.copyAny(pathValue.value()) : null; } - return DocumentCopies.copyAny(expression); + return false; } - private static Number evaluateGroupSumOperand(final Document source, final Object expression) { - final Object value = evaluateGroupExpression(source, expression); - if (value instanceof Number numberValue) { - return numberValue; + private static Object evaluateAccumulator( + final GroupAccumulator accumulator, final List sources) { + return switch (accumulator.operator()) { + case SUM -> accumulatorSum(sources, accumulator.expression()); + case AVG -> accumulatorAverage(sources, accumulator.expression()); + case MIN -> accumulatorMinMax(sources, accumulator.expression(), false); + case MAX -> accumulatorMinMax(sources, accumulator.expression(), true); + case FIRST -> sources.isEmpty() + ? null + : evaluateGroupExpression(sources.get(0), accumulator.expression()); + case LAST -> sources.isEmpty() + ? null + : evaluateGroupExpression(sources.get(sources.size() - 1), accumulator.expression()); + case PUSH -> accumulatorPush(sources, accumulator.expression(), false); + case ADD_TO_SET -> accumulatorPush(sources, accumulator.expression(), true); + case MERGE_OBJECTS -> accumulatorMergeObjects(sources, accumulator.expression()); + case PERCENTILE -> accumulatorPercentile(sources, accumulator.expression(), false); + case MEDIAN -> accumulatorPercentile(sources, accumulator.expression(), true); + case FIRST_N -> accumulatorPositionalN(sources, accumulator.expression(), false); + case LAST_N -> accumulatorPositionalN(sources, accumulator.expression(), true); + case MIN_N -> accumulatorMinMaxN(sources, accumulator.expression(), false); + case MAX_N -> accumulatorMinMaxN(sources, accumulator.expression(), true); + case TOP_N -> accumulatorTopBottomN(sources, accumulator.expression(), false); + case BOTTOM_N -> accumulatorTopBottomN(sources, accumulator.expression(), true); + }; + } + + private static Object accumulatorSum(final List sources, final Object expression) { + Number sum = 0L; + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, expression); + if (value instanceof Number number) { + sum = addNumbers(sum, number); + } } - return 0L; + return sum; } - private enum GroupAccumulatorOperator { - SUM, - FIRST, - ADD_TO_SET + private static Object accumulatorAverage(final List sources, final Object expression) { + double sum = 0d; + long count = 0L; + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, expression); + if (value instanceof Number number) { + sum += number.doubleValue(); + count++; + } + } + return count == 0 ? null : sum / count; } - private record GroupAccumulator(String outputField, GroupAccumulatorOperator operator, Object expression) { - private Number sumOperand(final Document source) { - return evaluateGroupSumOperand(source, expression); + private static Object accumulatorMinMax( + final List sources, final Object expression, final boolean maximum) { + Object selected = null; + boolean found = false; + for (final Document source : sources) { + final Object value = evaluateExpression(source, expression); + if (value == null || AggregationExpressions.isMissing(value)) { + continue; + } + if (!found) { + selected = value; + found = true; + continue; + } + final int comparison = MongoValueComparator.compare(value, selected); + if ((maximum && comparison > 0) || (!maximum && comparison < 0)) { + selected = value; + } } + return found ? DocumentCopies.copyAny(selected) : null; + } + + private static Object accumulatorPush( + final List sources, final Object expression, final boolean unique) { + final List output = new ArrayList<>(); + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, expression); + if (!unique || !containsByMongoEquality(output, value)) { + output.add(DocumentCopies.copyAny(value)); + } + } + return output; + } - private Object firstOperand(final Document source) { - return evaluateGroupExpression(source, expression); + private static Object accumulatorMergeObjects( + final List sources, final Object expression) { + final Document merged = new Document(); + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, expression); + if (value == null) { + continue; + } + if (!(value instanceof Map map)) { + throw new IllegalArgumentException("$mergeObjects accumulator requires document values"); + } + for (final Map.Entry entry : map.entrySet()) { + if (!(entry.getKey() instanceof String key)) { + throw new IllegalArgumentException("$mergeObjects document keys must be strings"); + } + merged.put(key, DocumentCopies.copyAny(entry.getValue())); + } + } + return merged; + } + + private static Object accumulatorPercentile( + final List sources, final Object rawDefinition, final boolean median) { + final Document definition = requireDocument( + rawDefinition, + median ? "$median requires a document" : "$percentile requires a document"); + final Set supported = median ? Set.of("input", "method") : Set.of("input", "p", "method"); + if (!supported.containsAll(definition.keySet()) + || !definition.containsKey("input") + || !definition.containsKey("method") + || (!median && !definition.containsKey("p"))) { + throw new IllegalArgumentException( + median + ? "$median requires input and method" + : "$percentile requires input, p, and method"); + } + if (!"approximate".equals(definition.get("method"))) { + throw new IllegalArgumentException("percentile method must be 'approximate'"); + } + + final List samples = new ArrayList<>(); + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, definition.get("input")); + if (value instanceof Number number && Double.isFinite(number.doubleValue())) { + samples.add(number.doubleValue()); + } + } + samples.sort(Double::compare); + if (median) { + return exactPercentile(samples, 0.5d); } - private Object addToSetOperand(final Document source) { - return evaluateGroupExpression(source, expression); + if (!(definition.get("p") instanceof List rawPercentiles) || rawPercentiles.isEmpty()) { + throw new IllegalArgumentException("$percentile.p must be a non-empty array"); + } + final List percentiles = new ArrayList<>(rawPercentiles.size()); + for (final Object rawPercentile : rawPercentiles) { + if (!(rawPercentile instanceof Number number) + || !Double.isFinite(number.doubleValue()) + || number.doubleValue() < 0d + || number.doubleValue() > 1d) { + throw new IllegalArgumentException("$percentile.p entries must be numbers in [0, 1]"); + } + percentiles.add(exactPercentile(samples, number.doubleValue())); } + return percentiles; } - private static Object evaluateGroupId(final Document source, final Object expression) { - if (expression instanceof String pathExpression && pathExpression.startsWith("$")) { - final PathValue pathValue = resolvePath(source, pathExpression.substring(1)); - return pathValue.present() ? DocumentCopies.copyAny(pathValue.value()) : null; + private static Object exactPercentile(final List sorted, final double percentile) { + if (sorted.isEmpty()) { + return null; + } + if (sorted.size() == 1) { + return sorted.get(0); } - return DocumentCopies.copyAny(expression); + final double rank = percentile * (sorted.size() - 1d); + final int lower = (int) Math.floor(rank); + final int upper = (int) Math.ceil(rank); + if (lower == upper) { + return sorted.get(lower); + } + final double fraction = rank - lower; + return sorted.get(lower) + (sorted.get(upper) - sorted.get(lower)) * fraction; } - private static Number addNumbers(final Object currentValue, final Number increment) { - if (!(currentValue instanceof Number currentNumber)) { - return normalizeNumber(increment.doubleValue()); + private static Object accumulatorPositionalN( + final List sources, final Object rawDefinition, final boolean fromEnd) { + final Document definition = requireNDefinition(rawDefinition, fromEnd ? "$lastN" : "$firstN", Set.of("input", "n")); + final int n = evaluateN(definition.get("n"), sources, fromEnd ? "$lastN" : "$firstN"); + final int start = fromEnd ? Math.max(0, sources.size() - n) : 0; + final int end = fromEnd ? sources.size() : Math.min(sources.size(), n); + final List output = new ArrayList<>(Math.max(0, end - start)); + for (int index = start; index < end; index++) { + output.add(DocumentCopies.copyAny(evaluateGroupExpression(sources.get(index), definition.get("input")))); } - return normalizeNumber(currentNumber.doubleValue() + increment.doubleValue()); + return output; } - private static boolean containsByMongoEquality(final List values, final Object candidate) { - for (final Object value : values) { - if (Objects.deepEquals(value, candidate)) { - return true; + private static Object accumulatorMinMaxN( + final List sources, final Object rawDefinition, final boolean maximum) { + final String operator = maximum ? "$maxN" : "$minN"; + final Document definition = requireNDefinition(rawDefinition, operator, Set.of("input", "n")); + final int n = evaluateN(definition.get("n"), sources, operator); + final List values = new ArrayList<>(); + for (final Document source : sources) { + final Object value = evaluateGroupExpression(source, definition.get("input")); + if (value != null) { + values.add(value); } } - return false; + values.sort((left, right) -> maximum + ? MongoValueComparator.compare(right, left) + : MongoValueComparator.compare(left, right)); + return new ArrayList<>(values.subList(0, Math.min(n, values.size()))); + } + + private static Object accumulatorTopBottomN( + final List sources, final Object rawDefinition, final boolean bottom) { + final String operator = bottom ? "$bottomN" : "$topN"; + final Document definition = requireNDefinition(rawDefinition, operator, Set.of("output", "sortBy", "n")); + final Document sortBy = requireDocument(definition.get("sortBy"), operator + ".sortBy must be a document"); + if (sortBy.isEmpty()) { + throw new IllegalArgumentException(operator + ".sortBy must not be empty"); + } + final List sortKeys = new ArrayList<>(); + for (final Map.Entry entry : sortBy.entrySet()) { + final int direction = parseSortDirection(entry.getValue()); + if (direction != 1 && direction != -1) { + throw new IllegalArgumentException(operator + ".sortBy directions must be 1 or -1"); + } + sortKeys.add(new SortKey(entry.getKey(), direction)); + } + final int n = evaluateN(definition.get("n"), sources, operator); + final List sorted = new ArrayList<>(sources); + sorted.sort((left, right) -> compareSortDocuments( + left, right, sortKeys, CollationSupport.Config.simple())); + final List output = new ArrayList<>(); + final int start = bottom ? Math.max(0, sorted.size() - n) : 0; + final int end = bottom ? sorted.size() : Math.min(n, sorted.size()); + for (int index = start; index < end; index++) { + output.add(DocumentCopies.copyAny(evaluateGroupExpression(sorted.get(index), definition.get("output")))); + } + return output; + } + + private static Document requireNDefinition( + final Object rawDefinition, final String operator, final Set requiredFields) { + final Document definition = requireDocument(rawDefinition, operator + " requires a document"); + if (!definition.keySet().equals(requiredFields)) { + throw new IllegalArgumentException(operator + " requires exactly " + requiredFields); + } + return definition; + } + + private static int evaluateN( + final Object expression, final List sources, final String operator) { + final Document context = sources.isEmpty() ? new Document() : sources.get(0); + final Object value = evaluateGroupExpression(context, expression); + if (!(value instanceof Number number) + || !Double.isFinite(number.doubleValue()) + || Math.rint(number.doubleValue()) != number.doubleValue() + || number.doubleValue() < 1d + || number.doubleValue() > Integer.MAX_VALUE) { + throw new IllegalArgumentException(operator + ".n must evaluate to a positive integer"); + } + return number.intValue(); + } + + private static final class GroupBucket { + private final Object id; + private final List sources = new ArrayList<>(); + + private GroupBucket(final Object id) { + this.id = DocumentCopies.copyAny(id); + } + + private Object id() { + return id; + } + + private List sources() { + return sources; + } } private static Number normalizeNumber(final double value) { @@ -437,6 +676,186 @@ private static List applySort( return sorted; } + private static List applySetWindowFields( + final List input, + final Object stageDefinition, + final CollationSupport.Config collation) { + final Document definition = + requireDocument(stageDefinition, "$setWindowFields stage requires a document"); + if (!Set.of("partitionBy", "sortBy", "output").containsAll(definition.keySet()) + || !definition.containsKey("output")) { + throw new IllegalArgumentException( + "$setWindowFields supports partitionBy, sortBy, and requires output"); + } + final Document outputDefinition = + requireDocument(definition.get("output"), "$setWindowFields.output must be a document"); + if (outputDefinition.isEmpty()) { + throw new IllegalArgumentException("$setWindowFields.output must not be empty"); + } + + final List sortKeys = new ArrayList<>(); + if (definition.containsKey("sortBy")) { + final Document sortDefinition = + requireDocument(definition.get("sortBy"), "$setWindowFields.sortBy must be a document"); + if (sortDefinition.isEmpty()) { + throw new IllegalArgumentException("$setWindowFields.sortBy must not be empty"); + } + for (final Map.Entry entry : sortDefinition.entrySet()) { + final int direction = parseSortDirection(entry.getValue()); + if (direction != 1 && direction != -1) { + throw new IllegalArgumentException("$setWindowFields.sortBy directions must be 1 or -1"); + } + sortKeys.add(new SortKey(entry.getKey(), direction)); + } + } + + final Map> partitions = new LinkedHashMap<>(); + for (final Document source : input) { + final Object partitionValue = definition.containsKey("partitionBy") + ? AggregationExpressions.nullIfMissing(evaluateExpression(source, definition.get("partitionBy"))) + : null; + partitions.computeIfAbsent(new GroupKey(partitionValue), ignored -> new ArrayList<>()).add(source); + } + + final List output = new ArrayList<>(input.size()); + for (final List partition : partitions.values()) { + if (!sortKeys.isEmpty()) { + partition.sort((left, right) -> compareSortDocuments(left, right, sortKeys, collation)); + } + applyWindowOutputs(partition, outputDefinition, sortKeys, collation, output); + } + return List.copyOf(output); + } + + private static void applyWindowOutputs( + final List partition, + final Document outputDefinition, + final List sortKeys, + final CollationSupport.Config collation, + final List output) { + long rank = 1L; + long denseRank = 1L; + for (int index = 0; index < partition.size(); index++) { + if (index > 0 && !sortKeys.isEmpty()) { + final boolean changed = compareSortDocuments( + partition.get(index - 1), partition.get(index), sortKeys, collation) + != 0; + if (changed) { + rank = index + 1L; + denseRank++; + } + } + + final Document expanded = DocumentCopies.copy(partition.get(index)); + for (final Map.Entry outputEntry : outputDefinition.entrySet()) { + final String outputPath = requireText(outputEntry.getKey(), "$setWindowFields output field"); + final Document operatorDefinition = requireDocument( + outputEntry.getValue(), "$setWindowFields output definitions must be documents"); + if (operatorDefinition.size() != 1) { + throw new IllegalArgumentException( + "$setWindowFields output definition must contain exactly one operator"); + } + final String operator = operatorDefinition.keySet().iterator().next(); + final Object value = switch (operator) { + case "$shift" -> evaluateShift( + partition, index, operatorDefinition.get(operator), sortKeys); + case "$documentNumber" -> { + requireEmptyWindowOperator(operatorDefinition.get(operator), "$documentNumber"); + if (sortKeys.isEmpty()) { + throw new IllegalArgumentException("$documentNumber requires sortBy"); + } + yield index + 1L; + } + case "$rank" -> { + requireEmptyWindowOperator(operatorDefinition.get(operator), "$rank"); + if (sortKeys.isEmpty()) { + throw new IllegalArgumentException("$rank requires sortBy"); + } + yield rank; + } + case "$denseRank" -> { + requireEmptyWindowOperator(operatorDefinition.get(operator), "$denseRank"); + if (sortKeys.isEmpty()) { + throw new IllegalArgumentException("$denseRank requires sortBy"); + } + yield denseRank; + } + default -> throw new UnsupportedFeatureException( + "aggregation.setWindowFields.operator." + operator, + "unsupported $setWindowFields operator: " + operator); + }; + if (AggregationExpressions.isMissing(value)) { + removePath(expanded, outputPath); + } else { + setPath(expanded, outputPath, value); + } + } + output.add(expanded); + } + } + + private static Object evaluateShift( + final List partition, + final int index, + final Object rawDefinition, + final List sortKeys) { + if (sortKeys.isEmpty()) { + throw new IllegalArgumentException("$shift requires sortBy"); + } + final Document definition = requireDocument(rawDefinition, "$shift requires a document"); + if (!Set.of("output", "by", "default").containsAll(definition.keySet()) + || !definition.containsKey("output") + || !definition.containsKey("by")) { + throw new IllegalArgumentException("$shift requires output and by, with optional default"); + } + final Object byValue = definition.get("by"); + if (!(byValue instanceof Number number) + || !Double.isFinite(number.doubleValue()) + || Math.rint(number.doubleValue()) != number.doubleValue() + || number.doubleValue() < Integer.MIN_VALUE + || number.doubleValue() > Integer.MAX_VALUE) { + throw new IllegalArgumentException("$shift.by must be a non-zero integer constant"); + } + final int targetIndex = index + number.intValue(); + if (targetIndex < 0 || targetIndex >= partition.size()) { + if (!definition.containsKey("default")) { + return null; + } + if (containsFieldReference(definition.get("default"))) { + throw new IllegalArgumentException("$shift.default must be a constant expression"); + } + return evaluateExpression(partition.get(index), definition.get("default")); + } + return evaluateExpression(partition.get(targetIndex), definition.get("output")); + } + + private static void requireEmptyWindowOperator(final Object value, final String operator) { + if (!(value instanceof Map map) || !map.isEmpty()) { + throw new IllegalArgumentException(operator + " requires an empty document"); + } + } + + private static boolean containsFieldReference(final Object value) { + if (value instanceof String stringValue) { + return stringValue.startsWith("$") && !stringValue.startsWith("$$"); + } + if (value instanceof Map map) { + for (final Object nested : map.values()) { + if (containsFieldReference(nested)) { + return true; + } + } + } + if (value instanceof List list) { + for (final Object nested : list) { + if (containsFieldReference(nested)) { + return true; + } + } + } + return false; + } + private static int parseSortDirection(final Object value) { if (!(value instanceof Number numberValue)) { throw new IllegalArgumentException("$sort directions must be numeric"); @@ -467,37 +886,9 @@ private static int compareSortDocuments( return 0; } - @SuppressWarnings({"rawtypes", "unchecked"}) private static int compareSortValues( final Object left, final Object right, final CollationSupport.Config collation) { - if (left == right) { - return 0; - } - if (left == null) { - return -1; - } - if (right == null) { - return 1; - } - if (left instanceof Number leftNumber && right instanceof Number rightNumber) { - return Double.compare(leftNumber.doubleValue(), rightNumber.doubleValue()); - } - if (left instanceof String leftString && right instanceof String rightString) { - return collation.compareStrings(leftString, rightString); - } - if (left instanceof Boolean leftBoolean && right instanceof Boolean rightBoolean) { - return Boolean.compare(leftBoolean, rightBoolean); - } - if (left.getClass().equals(right.getClass()) && left instanceof Comparable leftComparable) { - return leftComparable.compareTo(right); - } - - final int leftRank = sortTypeRank(left); - final int rightRank = sortTypeRank(right); - if (leftRank != rightRank) { - return Integer.compare(leftRank, rightRank); - } - return stableSortValue(left).compareTo(stableSortValue(right)); + return MongoValueComparator.compare(left, right, collation); } private static int sortTypeRank(final Object value) { @@ -694,7 +1085,12 @@ private static List applyFieldAssignmentStage( final Document expanded = DocumentCopies.copy(source); for (final Map.Entry entry : assignments.entrySet()) { final String fieldName = requireText(entry.getKey(), stageName + " field"); - setPath(expanded, fieldName, evaluateExpression(source, entry.getValue())); + final Object value = evaluateExpression(source, entry.getValue()); + if (AggregationExpressions.isMissing(value)) { + removePath(expanded, fieldName); + } else { + setPath(expanded, fieldName, value); + } } output.add(expanded); } @@ -756,7 +1152,8 @@ private static List applySortByCount( final CollationSupport.Config collation) { final Map buckets = new LinkedHashMap<>(); for (final Document source : input) { - final Object bucketValue = evaluateExpression(source, stageDefinition); + final Object bucketValue = AggregationExpressions.nullIfMissing( + evaluateExpression(source, stageDefinition)); final GroupKey bucketKey = new GroupKey(bucketValue); final CountBucket existing = buckets.get(bucketKey); if (existing == null) { @@ -1032,14 +1429,7 @@ private static int parseGraphLookupMaxDepth(final Object value) { } private static Object evaluateExpression(final Document source, final Object expression) { - if (expression instanceof String pathExpression && pathExpression.startsWith("$")) { - if ("$$ROOT".equals(pathExpression) || "$$CURRENT".equals(pathExpression)) { - return DocumentCopies.copy(source); - } - final PathValue pathValue = resolvePath(source, pathExpression.substring(1)); - return pathValue.present() ? DocumentCopies.copyAny(pathValue.value()) : null; - } - return DocumentCopies.copyAny(expression); + return AggregationExpressions.evaluate(source, expression); } private static PathValue resolvePath(final Object source, final String path) { @@ -1311,7 +1701,7 @@ private static final class GroupKey { private GroupKey(final Object value) { this.value = DocumentCopies.copyAny(value); - this.hashCode = Arrays.deepHashCode(new Object[] {this.value}); + this.hashCode = MongoValueComparator.hash(this.value); } @Override @@ -1322,7 +1712,7 @@ public boolean equals(final Object other) { if (!(other instanceof GroupKey that)) { return false; } - return Objects.deepEquals(value, that.value); + return MongoValueComparator.equals(value, that.value); } @Override diff --git a/src/main/java/org/jongodb/engine/CollectionStore.java b/src/main/java/org/jongodb/engine/CollectionStore.java index dc256a2..13b4dc8 100644 --- a/src/main/java/org/jongodb/engine/CollectionStore.java +++ b/src/main/java/org/jongodb/engine/CollectionStore.java @@ -51,6 +51,14 @@ default UpdateManyResult updateMany(final Document filter, final Document update return update(filter, update, true, false); } + default UpdateManyResult updatePipeline( + final Document filter, + final List pipeline, + final boolean multi, + final boolean upsert) { + throw new IllegalArgumentException("aggregation-pipeline updates are not supported yet"); + } + DeleteManyResult deleteMany(Document filter); record IndexDefinition( diff --git a/src/main/java/org/jongodb/engine/InMemoryCollectionStore.java b/src/main/java/org/jongodb/engine/InMemoryCollectionStore.java index 88ebc3c..9e51dde 100644 --- a/src/main/java/org/jongodb/engine/InMemoryCollectionStore.java +++ b/src/main/java/org/jongodb/engine/InMemoryCollectionStore.java @@ -351,6 +351,103 @@ public synchronized UpdateManyResult update( return new UpdateManyResult(matchedDocuments.size(), modifiedCount); } + @Override + public synchronized UpdateManyResult updatePipeline( + final Document filter, + final List pipeline, + final boolean multi, + final boolean upsert) { + pruneExpiredDocuments(); + final Document effectiveFilter = filter == null ? new Document() : DocumentCopies.copy(filter); + final List effectivePipeline = new ArrayList<>(); + for (final Document stage : Objects.requireNonNull(pipeline, "pipeline")) { + effectivePipeline.add(DocumentCopies.copy(Objects.requireNonNull(stage, "pipeline stage"))); + } + if (effectivePipeline.isEmpty()) { + throw new IllegalArgumentException("update pipeline must not be empty"); + } + + final List matchedDocuments = new ArrayList<>(); + for (final Document document : documents) { + if (QueryMatcher.matches(document, effectiveFilter)) { + matchedDocuments.add(document); + if (!multi) { + break; + } + } + } + + if (matchedDocuments.isEmpty()) { + if (!upsert) { + return new UpdateManyResult(0, 0); + } + return applyPipelineUpsert(effectiveFilter, effectivePipeline); + } + + final IdentityHashMap previewsByDocument = + new IdentityHashMap<>(matchedDocuments.size()); + long modifiedCount = 0L; + for (final Document document : matchedDocuments) { + final Document updated = applyPipelineToDocument(document, effectivePipeline); + final boolean modified = !Objects.deepEquals(document, updated); + previewsByDocument.put(document, new UpdatePreview(updated, modified)); + if (modified) { + modifiedCount++; + } + } + + if (modifiedCount > 0) { + final List candidateDocuments = new ArrayList<>(documents.size()); + for (final Document document : documents) { + final UpdatePreview preview = previewsByDocument.get(document); + candidateDocuments.add(preview == null || !preview.modified() + ? document + : preview.updatedDocument()); + } + validateUniqueConstraints(candidateDocuments, indexesByName.values()); + } + + for (final Document document : matchedDocuments) { + final UpdatePreview preview = previewsByDocument.get(document); + if (preview == null || !preview.modified()) { + continue; + } + document.clear(); + document.putAll(preview.updatedDocument()); + } + return new UpdateManyResult(matchedDocuments.size(), modifiedCount); + } + + private static Document applyPipelineToDocument( + final Document source, final List pipeline) { + final List results = AggregationPipeline.execute(List.of(DocumentCopies.copy(source)), pipeline); + if (results.size() != 1) { + throw new IllegalArgumentException("update pipeline must produce exactly one document per input document"); + } + final Document updated = results.get(0); + if (source.containsKey("_id") + && (!updated.containsKey("_id") + || !Objects.deepEquals(source.get("_id"), updated.get("_id")))) { + throw new IllegalArgumentException("update pipeline cannot change immutable field '_id'"); + } + return DocumentCopies.copy(updated); + } + + private UpdateManyResult applyPipelineUpsert( + final Document filter, final List pipeline) { + final Document upsertedDocument = applyPipelineToDocument(upsertSeed(filter), pipeline); + if (!upsertedDocument.containsKey("_id")) { + upsertedDocument.put("_id", new ObjectId()); + } + + final List candidateDocuments = new ArrayList<>(documents.size() + 1); + candidateDocuments.addAll(documents); + candidateDocuments.add(upsertedDocument); + validateUniqueConstraints(candidateDocuments, indexesByName.values()); + documents.add(upsertedDocument); + return new UpdateManyResult(0, 0, DocumentCopies.copyAny(upsertedDocument.get("_id"))); + } + private static List copyArrayFilters(final List arrayFilters) { if (arrayFilters == null || arrayFilters.isEmpty()) { return List.of(); diff --git a/src/main/java/org/jongodb/engine/MongoValueComparator.java b/src/main/java/org/jongodb/engine/MongoValueComparator.java new file mode 100644 index 0000000..0799b10 --- /dev/null +++ b/src/main/java/org/jongodb/engine/MongoValueComparator.java @@ -0,0 +1,305 @@ +package org.jongodb.engine; + +import java.lang.reflect.Array; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Date; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; +import org.bson.BsonRegularExpression; +import org.bson.BsonTimestamp; +import org.bson.types.Binary; +import org.bson.types.Decimal128; +import org.bson.types.MaxKey; +import org.bson.types.MinKey; +import org.bson.types.ObjectId; + +/** MongoDB-style BSON value ordering shared by update and aggregation operators. */ +final class MongoValueComparator { + private MongoValueComparator() {} + + static int compare(final Object left, final Object right) { + return compare(left, right, CollationSupport.Config.simple()); + } + + static int compare( + final Object left, + final Object right, + final CollationSupport.Config collation) { + if (left == right) { + return 0; + } + + final int leftRank = typeRank(left); + final int rightRank = typeRank(right); + if (leftRank != rightRank) { + return Integer.compare(leftRank, rightRank); + } + + if (left == null || left instanceof MinKey || left instanceof MaxKey) { + return 0; + } + if (left instanceof Number leftNumber && right instanceof Number rightNumber) { + return compareNumbers(leftNumber, rightNumber); + } + if (left instanceof CharSequence leftText && right instanceof CharSequence rightText) { + return collation.compareStrings(leftText.toString(), rightText.toString()); + } + if (left instanceof Map leftMap && right instanceof Map rightMap) { + return compareMaps(leftMap, rightMap, collation); + } + if (isArrayLike(left) && isArrayLike(right)) { + return compareLists(asList(left), asList(right), collation); + } + if (isBinary(left) && isBinary(right)) { + return compareBytes(binaryBytes(left), binaryBytes(right)); + } + if (left instanceof ObjectId leftObjectId && right instanceof ObjectId rightObjectId) { + return compareBytes(leftObjectId.toByteArray(), rightObjectId.toByteArray()); + } + if (left instanceof Boolean leftBoolean && right instanceof Boolean rightBoolean) { + return Boolean.compare(leftBoolean, rightBoolean); + } + if (isDateLike(left) && isDateLike(right)) { + return Long.compare(dateMillis(left), dateMillis(right)); + } + if (left instanceof BsonTimestamp leftTimestamp && right instanceof BsonTimestamp rightTimestamp) { + final int timeComparison = Integer.compare(leftTimestamp.getTime(), rightTimestamp.getTime()); + return timeComparison != 0 + ? timeComparison + : Integer.compare(leftTimestamp.getInc(), rightTimestamp.getInc()); + } + if (isRegex(left) && isRegex(right)) { + return regexValue(left).compareTo(regexValue(right)); + } + + if (left.getClass().equals(right.getClass()) && left instanceof Comparable comparable) { + @SuppressWarnings("unchecked") + final Comparable typed = (Comparable) comparable; + return typed.compareTo(right); + } + return stableValue(left).compareTo(stableValue(right)); + } + + static boolean equals(final Object left, final Object right) { + return typeRank(left) == typeRank(right) && compare(left, right) == 0; + } + + static int hash(final Object value) { + if (value == null) { + return 0; + } + if (value instanceof Number number) { + if (isSpecialFloating(number)) { + return Double.hashCode(number.doubleValue()); + } + try { + return toBigDecimal(number).stripTrailingZeros().hashCode(); + } catch (final ArithmeticException | NumberFormatException ignored) { + return Double.hashCode(number.doubleValue()); + } + } + if (value instanceof Map map) { + int hash = 1; + for (final Map.Entry entry : map.entrySet()) { + hash = 31 * hash + String.valueOf(entry.getKey()).hashCode(); + hash = 31 * hash + hash(entry.getValue()); + } + return hash; + } + if (isArrayLike(value)) { + int hash = 1; + for (final Object item : asList(value)) { + hash = 31 * hash + hash(item); + } + return hash; + } + if (isDateLike(value)) { + return Long.hashCode(dateMillis(value)); + } + if (isBinary(value)) { + return java.util.Arrays.hashCode(binaryBytes(value)); + } + return value.hashCode(); + } + + private static int compareNumbers(final Number left, final Number right) { + if (isSpecialFloating(left) || isSpecialFloating(right)) { + return Double.compare(left.doubleValue(), right.doubleValue()); + } + try { + return toBigDecimal(left).compareTo(toBigDecimal(right)); + } catch (final ArithmeticException | NumberFormatException ignored) { + return Double.compare(left.doubleValue(), right.doubleValue()); + } + } + + private static BigDecimal toBigDecimal(final Number value) { + if (value instanceof BigDecimal bigDecimal) { + return bigDecimal; + } + if (value instanceof BigInteger bigInteger) { + return new BigDecimal(bigInteger); + } + if (value instanceof Decimal128 decimal128) { + return decimal128.bigDecimalValue(); + } + if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long) { + return BigDecimal.valueOf(value.longValue()); + } + if (value instanceof Float || value instanceof Double) { + return BigDecimal.valueOf(value.doubleValue()); + } + return new BigDecimal(value.toString()); + } + + private static boolean isSpecialFloating(final Number value) { + return (value instanceof Double doubleValue && !Double.isFinite(doubleValue)) + || (value instanceof Float floatValue && !Float.isFinite(floatValue)); + } + + private static int compareMaps( + final Map left, + final Map right, + final CollationSupport.Config collation) { + final Iterator> leftEntries = left.entrySet().iterator(); + final Iterator> rightEntries = right.entrySet().iterator(); + while (leftEntries.hasNext() && rightEntries.hasNext()) { + final Map.Entry leftEntry = leftEntries.next(); + final Map.Entry rightEntry = rightEntries.next(); + final int typeComparison = Integer.compare(typeRank(leftEntry.getValue()), typeRank(rightEntry.getValue())); + if (typeComparison != 0) { + return typeComparison; + } + final int keyComparison = String.valueOf(leftEntry.getKey()).compareTo(String.valueOf(rightEntry.getKey())); + if (keyComparison != 0) { + return keyComparison; + } + final int valueComparison = compare(leftEntry.getValue(), rightEntry.getValue(), collation); + if (valueComparison != 0) { + return valueComparison; + } + } + return Boolean.compare(leftEntries.hasNext(), rightEntries.hasNext()); + } + + private static int compareLists( + final List left, + final List right, + final CollationSupport.Config collation) { + final int commonSize = Math.min(left.size(), right.size()); + for (int index = 0; index < commonSize; index++) { + final int comparison = compare(left.get(index), right.get(index), collation); + if (comparison != 0) { + return comparison; + } + } + return Integer.compare(left.size(), right.size()); + } + + private static int compareBytes(final byte[] left, final byte[] right) { + final int commonLength = Math.min(left.length, right.length); + for (int index = 0; index < commonLength; index++) { + final int comparison = Integer.compare(Byte.toUnsignedInt(left[index]), Byte.toUnsignedInt(right[index])); + if (comparison != 0) { + return comparison; + } + } + return Integer.compare(left.length, right.length); + } + + private static boolean isArrayLike(final Object value) { + return value instanceof List || (value != null && value.getClass().isArray() && !(value instanceof byte[])); + } + + private static List asList(final Object value) { + if (value instanceof List list) { + return new ArrayList<>(list); + } + final int length = Array.getLength(value); + final List output = new ArrayList<>(length); + for (int index = 0; index < length; index++) { + output.add(Array.get(value, index)); + } + return output; + } + + private static boolean isBinary(final Object value) { + return value instanceof Binary || value instanceof byte[]; + } + + private static byte[] binaryBytes(final Object value) { + return value instanceof Binary binary ? binary.getData() : (byte[]) value; + } + + private static boolean isDateLike(final Object value) { + return value instanceof Date || value instanceof Instant; + } + + private static long dateMillis(final Object value) { + return value instanceof Date date ? date.getTime() : ((Instant) value).toEpochMilli(); + } + + private static boolean isRegex(final Object value) { + return value instanceof Pattern || value instanceof BsonRegularExpression; + } + + private static String regexValue(final Object value) { + if (value instanceof Pattern pattern) { + return pattern.pattern() + '\u0000' + pattern.flags(); + } + final BsonRegularExpression expression = (BsonRegularExpression) value; + return expression.getPattern() + '\u0000' + expression.getOptions(); + } + + private static int typeRank(final Object value) { + if (value instanceof MinKey) { + return 0; + } + if (value == null) { + return 1; + } + if (value instanceof Number) { + return 2; + } + if (value instanceof CharSequence || value instanceof Character) { + return 3; + } + if (value instanceof Map) { + return 4; + } + if (isArrayLike(value)) { + return 5; + } + if (isBinary(value)) { + return 6; + } + if (value instanceof ObjectId) { + return 7; + } + if (value instanceof Boolean) { + return 8; + } + if (isDateLike(value)) { + return 9; + } + if (value instanceof BsonTimestamp) { + return 10; + } + if (isRegex(value)) { + return 11; + } + if (value instanceof MaxKey) { + return 100; + } + return 50; + } + + private static String stableValue(final Object value) { + return value.getClass().getName() + ':' + value; + } +} diff --git a/src/main/java/org/jongodb/engine/UpdateApplier.java b/src/main/java/org/jongodb/engine/UpdateApplier.java index c944711..2b14d68 100644 --- a/src/main/java/org/jongodb/engine/UpdateApplier.java +++ b/src/main/java/org/jongodb/engine/UpdateApplier.java @@ -43,12 +43,15 @@ static ParsedUpdate parse(final Document update, final List arrayFilte final List setOperations = new ArrayList<>(); final List setOnInsertOperations = new ArrayList<>(); final List incrementOperations = new ArrayList<>(); + final List minMaxOperations = new ArrayList<>(); final List unsetOperations = new ArrayList<>(); final List addToSetOperations = new ArrayList<>(); + final List claimedPaths = new ArrayList<>(); for (final Map.Entry entry : update.entrySet()) { final String operator = entry.getKey(); final Map definition = readDefinition(operator, entry.getValue(), parsedArrayFilters); + validateNoConflictingPaths(definition.keySet(), claimedPaths); switch (operator) { case "$set": for (final Map.Entry setEntry : definition.entrySet()) { @@ -71,6 +74,15 @@ static ParsedUpdate parse(final Document update, final List arrayFilte incrementOperations.add(new IncOperation(incEntry.getKey(), (Number) delta)); } break; + case "$min": + case "$max": + for (final Map.Entry minMaxEntry : definition.entrySet()) { + minMaxOperations.add(new MinMaxOperation( + minMaxEntry.getKey(), + minMaxEntry.getValue(), + "$max".equals(operator))); + } + break; case "$unset": unsetOperations.addAll(definition.keySet()); break; @@ -91,11 +103,27 @@ static ParsedUpdate parse(final Document update, final List arrayFilte setOperations, setOnInsertOperations, incrementOperations, + minMaxOperations, unsetOperations, addToSetOperations, parsedArrayFilters); } + private static void validateNoConflictingPaths( + final Set newPaths, final List claimedPaths) { + for (final String path : newPaths) { + for (final String claimedPath : claimedPaths) { + if (path.equals(claimedPath) + || path.startsWith(claimedPath + ".") + || claimedPath.startsWith(path + ".")) { + throw new IllegalArgumentException( + "updating the path '" + path + "' would create a conflict at '" + claimedPath + "'"); + } + } + claimedPaths.add(path); + } + } + static void validateApplicable(final Document document, final ParsedUpdate update) { validateApplicable(document, update, false); } @@ -131,6 +159,9 @@ private static void validateApplicable( "$inc target for '" + operation.path() + "' must be numeric"); } } + for (final MinMaxOperation operation : update.minMaxOperations()) { + ensureWritablePath(document, operation.path(), update.arrayFilterBindings()); + } for (final AddToSetOperation operation : update.addToSetOperations()) { ensureWritablePath(document, operation.path(), update.arrayFilterBindings()); @@ -172,6 +203,9 @@ private static boolean apply( for (final IncOperation operation : update.incrementOperations()) { modified |= applyIncrement(document, operation.path(), operation.delta()); } + for (final MinMaxOperation operation : update.minMaxOperations()) { + modified |= applyMinMax(document, operation); + } for (final String path : update.unsetOperations()) { modified |= applyUnset(document, path, update.arrayFilterBindings()); } @@ -402,6 +436,26 @@ private static boolean applyIncrement(final Document document, final String path return true; } + private static boolean applyMinMax(final Document document, final MinMaxOperation operation) { + final Map parent = getOrCreateParent(document, operation.path()); + final String leaf = leaf(operation.path()); + final Object candidate = DocumentCopies.copyAny(operation.value()); + + if (!parent.containsKey(leaf)) { + parent.put(leaf, candidate); + return true; + } + + final Object current = parent.get(leaf); + final int comparison = MongoValueComparator.compare(candidate, current); + final boolean replace = operation.maximum() ? comparison > 0 : comparison < 0; + if (!replace) { + return false; + } + parent.put(leaf, candidate); + return true; + } + private static boolean applyUnset( final Document document, final String path, final ArrayFilterBindings arrayFilterBindings) { if (pathContainsArrayFilter(path)) { @@ -521,7 +575,7 @@ private static List castList(final List source) { private static boolean containsByMongoEquality(final List values, final Object candidate) { for (final Object value : values) { - if (valueEquals(value, candidate)) { + if (MongoValueComparator.equals(value, candidate)) { return true; } } @@ -819,6 +873,7 @@ static final class ParsedUpdate { private final List setOperations; private final List setOnInsertOperations; private final List incrementOperations; + private final List minMaxOperations; private final List unsetOperations; private final List addToSetOperations; private final Document replacementDocument; @@ -828,6 +883,7 @@ private ParsedUpdate( final List setOperations, final List setOnInsertOperations, final List incrementOperations, + final List minMaxOperations, final List unsetOperations, final List addToSetOperations, final Document replacementDocument, @@ -835,6 +891,7 @@ private ParsedUpdate( this.setOperations = List.copyOf(setOperations); this.setOnInsertOperations = List.copyOf(setOnInsertOperations); this.incrementOperations = List.copyOf(incrementOperations); + this.minMaxOperations = List.copyOf(minMaxOperations); this.unsetOperations = List.copyOf(unsetOperations); this.addToSetOperations = List.copyOf(addToSetOperations); this.replacementDocument = replacementDocument == null ? null : DocumentCopies.copy(replacementDocument); @@ -845,6 +902,7 @@ static ParsedUpdate operator( final List setOperations, final List setOnInsertOperations, final List incrementOperations, + final List minMaxOperations, final List unsetOperations, final List addToSetOperations, final ArrayFilterBindings arrayFilterBindings) { @@ -852,6 +910,7 @@ static ParsedUpdate operator( setOperations, setOnInsertOperations, incrementOperations, + minMaxOperations, unsetOperations, addToSetOperations, null, @@ -865,6 +924,7 @@ static ParsedUpdate replacement(final Document replacementDocument) { List.of(), List.of(), List.of(), + List.of(), replacementDocument, ArrayFilterBindings.empty()); } @@ -889,6 +949,10 @@ List incrementOperations() { return Collections.unmodifiableList(incrementOperations); } + List minMaxOperations() { + return Collections.unmodifiableList(minMaxOperations); + } + List unsetOperations() { return Collections.unmodifiableList(unsetOperations); } @@ -987,6 +1051,8 @@ private record SetOnInsertOperation(String path, Object value) {} private record IncOperation(String path, Number delta) {} + private record MinMaxOperation(String path, Object value, boolean maximum) {} + private record AddToSetOperation(String path, List values) {} private record PathLookup(boolean exists, Object value) { diff --git a/src/test/java/org/jongodb/command/CommandDispatcherE2ETest.java b/src/test/java/org/jongodb/command/CommandDispatcherE2ETest.java index b2109eb..525ba33 100644 --- a/src/test/java/org/jongodb/command/CommandDispatcherE2ETest.java +++ b/src/test/java/org/jongodb/command/CommandDispatcherE2ETest.java @@ -882,6 +882,104 @@ void updateCommandSupportsPipelineSetUnsetSubset() { assertEquals("seoul", updated.getDocument("profile").getString("city").getValue()); } + @Test + void updateCommandSupportsMinMaxWithUpsertNumbersDatesAndInc() { + final CommandDispatcher dispatcher = new CommandDispatcher(new EngineBackedCommandStore(new InMemoryEngineStore())); + + final BsonDocument upsert = dispatcher.dispatch(BsonDocument.parse( + """ + { + "update": "rollups", + "$db": "app", + "updates": [{ + "q": {"_id": "bucket"}, + "u": { + "$inc": {"count": 3}, + "$max": {"durationMaxMs": 4000.0, "lastEventAt": {"$date": "2026-07-21T09:30:00Z"}}, + "$min": {"durationMinMs": 40.0} + }, + "upsert": true + }] + } + """)); + assertEquals(1.0d, upsert.get("ok").asNumber().doubleValue()); + assertEquals(1, upsert.getArray("upserted").size()); + + dispatcher.dispatch(BsonDocument.parse( + """ + { + "update": "rollups", + "$db": "app", + "updates": [{ + "q": {"_id": "bucket"}, + "u": { + "$inc": {"count": 2}, + "$max": {"durationMaxMs": 2000.0, "lastEventAt": {"$date": "2026-07-21T09:00:00Z"}}, + "$min": {"durationMinMs": 80} + } + }] + } + """)); + dispatcher.dispatch(BsonDocument.parse( + """ + {"update":"rollups","$db":"app","updates":[{ + "q":{"_id":"bucket"}, + "u":{"$max":{"durationMaxMs":5000.0,"lastEventAt":{"$date":"2026-07-21T10:00:00Z"}},"$min":{"durationMinMs":20}} + }]} + """)); + + final BsonDocument found = dispatcher.dispatch(BsonDocument.parse( + "{\"find\":\"rollups\",\"$db\":\"app\",\"filter\":{\"_id\":\"bucket\"}}")); + final BsonDocument rollup = found.getDocument("cursor").getArray("firstBatch").get(0).asDocument(); + assertEquals(5, rollup.getInt32("count").getValue()); + assertEquals(5000.0d, rollup.getDouble("durationMaxMs").getValue()); + assertEquals(20, rollup.getInt32("durationMinMs").getValue()); + assertEquals(1784628000000L, rollup.getDateTime("lastEventAt").getValue()); + } + + @Test + void updateCommandEvaluatesConditionalAggregationPipelineAtomically() { + final CommandDispatcher dispatcher = new CommandDispatcher(new EngineBackedCommandStore(new InMemoryEngineStore())); + final String updateTemplate = """ + { + "update": "activity", + "$db": "app", + "updates": [{ + "q": {"_id": "bucket|account"}, + "u": [{ + "$set": { + "lastEventAt": {"$max": ["$lastEventAt", {"$date": "%s"}]}, + "lastEvent": {"$cond": [ + {"$or": [ + {"$eq": [{"$type": "$lastEventAt"}, "missing"]}, + {"$lt": ["$lastEventAt", {"$date": "%s"}]} + ]}, + "%s", + "$lastEvent" + ]} + } + }, {"$replaceWith": {"$mergeObjects": ["$$ROOT", {"updated": true}]}}], + "upsert": true + }] + } + """; + + final BsonDocument inserted = dispatcher.dispatch(BsonDocument.parse( + updateTemplate.formatted("2026-07-21T09:30:00Z", "2026-07-21T09:30:00Z", "batch.run"))); + assertEquals(1.0d, inserted.get("ok").asNumber().doubleValue()); + dispatcher.dispatch(BsonDocument.parse( + updateTemplate.formatted("2026-07-21T09:00:00Z", "2026-07-21T09:00:00Z", "older.run"))); + dispatcher.dispatch(BsonDocument.parse( + updateTemplate.formatted("2026-07-21T10:00:00Z", "2026-07-21T10:00:00Z", "newer.run"))); + + final BsonDocument found = dispatcher.dispatch(BsonDocument.parse( + "{\"find\":\"activity\",\"$db\":\"app\",\"filter\":{\"_id\":\"bucket|account\"}}")); + final BsonDocument activity = found.getDocument("cursor").getArray("firstBatch").get(0).asDocument(); + assertEquals("newer.run", activity.getString("lastEvent").getValue()); + assertEquals(1784628000000L, activity.getDateTime("lastEventAt").getValue()); + assertTrue(activity.getBoolean("updated").getValue()); + } + @Test void updateCommandSupportsReplaceOneSemantics() { final CommandDispatcher dispatcher = new CommandDispatcher(new EngineBackedCommandStore(new InMemoryEngineStore())); @@ -1453,17 +1551,21 @@ void updateCommandRejectsInvalidPayloadShapes() { assertCommandError(arrayFilterWithUnsupportedOperator, "BadValue"); final BsonDocument unsupportedPipelineStage = dispatcher.dispatch(BsonDocument.parse( - "{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":[{\"$replaceRoot\":{\"newRoot\":{\"x\":1}}}]}]}")); + "{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":[{\"$match\":{\"x\":1}}]}]}")); assertCommandError(unsupportedPipelineStage, "BadValue"); - final BsonDocument pipelineExpressionNotSupported = dispatcher.dispatch(BsonDocument.parse( + final BsonDocument pipelineExpressionSupported = dispatcher.dispatch(BsonDocument.parse( "{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":[{\"$set\":{\"a\":\"$other\"}}]}]}")); - assertCommandError(pipelineExpressionNotSupported, "BadValue"); + assertEquals(1.0d, pipelineExpressionSupported.get("ok").asNumber().doubleValue()); final BsonDocument unsupportedPositionalPath = dispatcher.dispatch(BsonDocument.parse( "{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":{\"$set\":{\"items.$.qty\":1}}}]}")); assertCommandError(unsupportedPositionalPath, "BadValue"); + final BsonDocument conflictingUpdatePaths = dispatcher.dispatch(BsonDocument.parse( + "{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":{\"$set\":{\"stats\":1},\"$max\":{\"stats.maximum\":2}}}]}")); + assertCommandError(conflictingUpdatePaths, "BadValue"); + final BsonDocument setOnInsertTypeMismatch = dispatcher.dispatch( BsonDocument.parse("{\"update\":\"users\",\"updates\":[{\"q\":{},\"u\":{\"$setOnInsert\":1}}]}")); assertCommandError(setOnInsertTypeMismatch, "TypeMismatch"); diff --git a/src/test/java/org/jongodb/engine/AggregationPipelineTest.java b/src/test/java/org/jongodb/engine/AggregationPipelineTest.java index 9bcd3f6..f7ffe21 100644 --- a/src/test/java/org/jongodb/engine/AggregationPipelineTest.java +++ b/src/test/java/org/jongodb/engine/AggregationPipelineTest.java @@ -8,7 +8,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; import java.util.List; +import java.time.Instant; import org.bson.Document; import org.junit.jupiter.api.Test; @@ -101,4 +104,110 @@ void unwindRejectsInvalidIncludeArrayIndexName() { "$unwind", new Document("path", "$tags").append("includeArrayIndex", "$index"))))); } + + @Test + void groupSupportsCompoundIdsCoreAccumulatorsAndMergeObjects() { + final List source = List.of( + new Document("event", "run").append("tenant", "crown").append("duration", 10).append("labels", new Document("a", 1)), + new Document("event", "run").append("tenant", "crown").append("duration", 30.0d).append("labels", new Document("b", 2)), + new Document("event", "run").append("tenant", "crown").append("labels", new Document("a", 3))); + final Document group = new Document("_id", new Document("event", "$event").append("tenant", "$tenant")) + .append("count", new Document("$sum", 1)) + .append("total", new Document("$sum", "$duration")) + .append("average", new Document("$avg", "$duration")) + .append("minimum", new Document("$min", "$duration")) + .append("maximum", new Document("$max", "$duration")) + .append("first", new Document("$first", "$duration")) + .append("last", new Document("$last", "$duration")) + .append("all", new Document("$push", "$duration")) + .append("unique", new Document("$addToSet", "$duration")) + .append("labels", new Document("$mergeObjects", "$labels")); + + final Document result = AggregationPipeline.execute(source, List.of(new Document("$group", group))).get(0); + + assertEquals(new Document("event", "run").append("tenant", "crown"), result.get("_id")); + assertEquals(3L, result.get("count")); + assertEquals(40L, result.get("total")); + assertEquals(20.0d, result.get("average")); + assertEquals(10, result.get("minimum")); + assertEquals(30.0d, result.get("maximum")); + assertEquals(10, result.get("first")); + assertNull(result.get("last")); + assertEquals(Arrays.asList(10, 30.0d, null), result.get("all")); + assertEquals(new Document("a", 3).append("b", 2), result.get("labels")); + } + + @Test + void dateTruncSupportsUtcCalendarAndFixedUnits() { + final Date timestamp = Date.from(Instant.parse("2026-07-21T09:37:42.987Z")); + final Document projection = new Document("hour", new Document("$dateTrunc", new Document("date", "$ts").append("unit", "hour"))) + .append("day", new Document("$dateTrunc", new Document("date", "$ts").append("unit", "day").append("timezone", "UTC"))) + .append("week", new Document("$dateTrunc", new Document("date", "$ts").append("unit", "week").append("startOfWeek", "mon"))) + .append("month", new Document("$dateTrunc", new Document("date", "$ts").append("unit", "month"))); + + final Document result = AggregationPipeline.execute( + List.of(new Document("ts", timestamp)), List.of(new Document("$project", projection))) + .get(0); + + assertEquals(Date.from(Instant.parse("2026-07-21T09:00:00Z")), result.getDate("hour")); + assertEquals(Date.from(Instant.parse("2026-07-21T00:00:00Z")), result.getDate("day")); + assertEquals(Date.from(Instant.parse("2026-07-20T00:00:00Z")), result.getDate("week")); + assertEquals(Date.from(Instant.parse("2026-07-01T00:00:00Z")), result.getDate("month")); + } + + @Test + void groupSupportsPercentileMedianAndNAccumulators() { + final List source = List.of( + new Document("name", "A").append("score", 10), + new Document("name", "B").append("score", 30), + new Document("name", "C").append("score", 20), + new Document("name", "D").append("score", 40)); + final Document group = new Document("_id", null) + .append("p", new Document("$percentile", new Document("input", "$score").append("p", List.of(0.5d, 0.95d)).append("method", "approximate"))) + .append("med", new Document("$median", new Document("input", "$score").append("method", "approximate"))) + .append("first", new Document("$firstN", new Document("input", "$name").append("n", 2))) + .append("last", new Document("$lastN", new Document("input", "$name").append("n", 2))) + .append("min", new Document("$minN", new Document("input", "$score").append("n", 2))) + .append("max", new Document("$maxN", new Document("input", "$score").append("n", 2))) + .append("top", new Document("$topN", new Document("output", "$name").append("sortBy", new Document("score", -1)).append("n", 2))) + .append("bottom", new Document("$bottomN", new Document("output", "$name").append("sortBy", new Document("score", -1)).append("n", 2))); + + final Document result = AggregationPipeline.execute(source, List.of(new Document("$group", group))).get(0); + + assertEquals(List.of(25.0d, 38.5d), result.get("p")); + assertEquals(25.0d, result.get("med")); + assertEquals(List.of("A", "B"), result.get("first")); + assertEquals(List.of("C", "D"), result.get("last")); + assertEquals(List.of(10, 20), result.get("min")); + assertEquals(List.of(40, 30), result.get("max")); + assertEquals(List.of("D", "B"), result.get("top")); + assertEquals(List.of("C", "A"), result.get("bottom")); + } + + @Test + void setWindowFieldsSupportsShiftDocumentNumberAndRanks() { + final List source = List.of( + new Document("account", "a").append("at", 20).append("event", "second"), + new Document("account", "a").append("at", 10).append("event", "first"), + new Document("account", "b").append("at", 5).append("event", "only")); + final Document outputs = new Document("previous", new Document("$shift", new Document("output", "$event").append("by", -1).append("default", "none"))) + .append("number", new Document("$documentNumber", new Document())) + .append("rank", new Document("$rank", new Document())) + .append("denseRank", new Document("$denseRank", new Document())); + final Document stage = new Document("partitionBy", "$account") + .append("sortBy", new Document("at", 1)) + .append("output", outputs); + + final List result = AggregationPipeline.execute( + source, List.of(new Document("$setWindowFields", stage))); + + assertEquals("first", result.get(0).getString("event")); + assertEquals("none", result.get(0).getString("previous")); + assertEquals(1L, result.get(0).get("number")); + assertEquals("second", result.get(1).getString("event")); + assertEquals("first", result.get(1).getString("previous")); + assertEquals(2L, result.get(1).get("rank")); + assertEquals("only", result.get(2).getString("event")); + assertEquals(1L, result.get(2).get("denseRank")); + } } diff --git a/src/test/java/org/jongodb/engine/UpdateApplierTest.java b/src/test/java/org/jongodb/engine/UpdateApplierTest.java index d8e1223..3b50d6a 100644 --- a/src/test/java/org/jongodb/engine/UpdateApplierTest.java +++ b/src/test/java/org/jongodb/engine/UpdateApplierTest.java @@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.List; +import java.util.Date; import org.bson.Document; import org.junit.jupiter.api.Test; @@ -33,10 +34,10 @@ void applySupportsSetIncAndUnsetOperators() { @Test void applyReturnsFalseWhenUpdateDoesNotChangeDocument() { - Document target = new Document("count", 5); + Document target = new Document("count", 5).append("offset", 2); Document update = new Document("$set", new Document("count", 5)) - .append("$inc", new Document("count", 0)) + .append("$inc", new Document("offset", 0)) .append("$unset", new Document("missing", true)); UpdateApplier.ParsedUpdate parsed = UpdateApplier.parse(update); @@ -46,6 +47,53 @@ void applyReturnsFalseWhenUpdateDoesNotChangeDocument() { assertEquals(5, target.getInteger("count")); } + @Test + void applySupportsMinMaxForMissingMixedNumbersDatesAndDottedPaths() { + final Date older = new Date(1_000L); + final Date newer = new Date(2_000L); + final Document target = new Document("duration", 3_000) + .append("lastEventAt", older) + .append("stats", new Document("minimum", 8.0d)); + final Document update = new Document("$inc", new Document("count", 3L)) + .append("$max", new Document("duration", 4_000.0d).append("lastEventAt", newer)) + .append("$min", new Document("stats.minimum", 5).append("firstSeenAt", older)); + + final UpdateApplier.ParsedUpdate parsed = UpdateApplier.parse(update); + UpdateApplier.validateApplicable(target, parsed); + + assertTrue(UpdateApplier.apply(target, parsed)); + assertEquals(4_000.0d, target.getDouble("duration")); + assertEquals(newer, target.getDate("lastEventAt")); + assertEquals(5, target.get("stats", Document.class).getInteger("minimum")); + assertEquals(older, target.getDate("firstSeenAt")); + assertEquals(3L, target.getLong("count")); + + final Document noOp = new Document("$max", new Document("duration", 2_000.0d).append("lastEventAt", older)) + .append("$min", new Document("stats.minimum", 9.0d)); + assertFalse(UpdateApplier.apply(target, UpdateApplier.parse(noOp))); + } + + @Test + void applyForUpsertInsertCreatesMinAndMaxFields() { + final Document target = new Document("_id", "rollup"); + final Document update = new Document("$max", new Document("durationMaxMs", 4_000.0d)) + .append("$min", new Document("durationMinMs", 40.0d)); + + assertTrue(UpdateApplier.applyForUpsertInsert(target, UpdateApplier.parse(update))); + assertEquals(4_000.0d, target.getDouble("durationMaxMs")); + assertEquals(40.0d, target.getDouble("durationMinMs")); + } + + @Test + void parseRejectsConflictingUpdatePathsAcrossOperators() { + final IllegalArgumentException error = assertThrows( + IllegalArgumentException.class, + () -> UpdateApplier.parse(new Document("$set", new Document("stats", 1)) + .append("$max", new Document("stats.maximum", 2)))); + + assertTrue(error.getMessage().contains("would create a conflict")); + } + @Test void validateApplicableRejectsNonNumericIncTarget() { Document target = new Document("count", "oops");