@Override
    public void upgrade() {
        if (clusterConfigService.get(MigrationCompleted.class) != null) {
            LOG.debug("Migration already completed!");
            return;
        }

        final List<SearchPivotLimitMigration> pivotLimitMigrations = StreamSupport.stream(this.searches.find().spliterator(), false)
                .flatMap(document -> {
                    final String searchId = document.get("_id", ObjectId.class).toHexString();
                    final List<Document> queries = document.get("queries", Collections.emptyList());
                    return EntryStream.of(queries)
                            .flatMap(entry -> {
                                final Integer queryIndex = entry.getKey();
                                final List<Document> searchTypes = entry.getValue().get("search_types", Collections.emptyList());
                                return EntryStream.of(searchTypes)
                                        .filter(searchType -> "pivot".equals(searchType.getValue().getString("type")))
                                        .flatMap(searchTypeEntry -> {
                                            final Document searchType = searchTypeEntry.getValue();
                                            final Integer searchTypeIndex = searchTypeEntry.getKey();
                                            final boolean hasRowLimit = searchType.containsKey("row_limit");
                                            final boolean hasColumnLimit = searchType.containsKey("column_limit");
                                            final Optional<Integer> rowLimit = Optional.ofNullable(searchType.getInteger("row_limit"));
                                            final Optional<Integer> columnLimit = Optional.ofNullable(searchType.getInteger("column_limit"));

                                            if (searchTypeIndex != null && (hasRowLimit || hasColumnLimit)) {
                                                return Stream.of(new SearchPivotLimitMigration(searchId, queryIndex, searchTypeIndex, rowLimit, columnLimit));
                                            }
                                            return Stream.empty();
                                        });
                            });
                })
                .collect(Collectors.toList());

        final List<WriteModel<Document>> operations = pivotLimitMigrations.stream()
                .flatMap(pivotMigration -> {
                    final ImmutableList.Builder<WriteModel<Document>> builder = ImmutableList.builder();
                    builder.add(
                            updateSearch(
                                    pivotMigration.searchId(),
                                    doc("$unset", doc(pivotPath(pivotMigration) + ".row_limit", 1))
                            )
                    );
                    builder.add(
                            updateSearch(
                                    pivotMigration.searchId(),
                                    doc("$set", doc(pivotPath(pivotMigration) + ".row_groups.$[pivot].limit", pivotMigration.rowLimit.orElse(DEFAULT_LIMIT))),
                                    matchValuePivots
                            )
                    );
                    builder.add(
                            updateSearch(
                                    pivotMigration.searchId(),
                                    doc("$unset", doc(pivotPath(pivotMigration) + ".column_limit", 1))
                            )
                    );
                    builder.add(
                            updateSearch(
                                    pivotMigration.searchId(),
                                    doc("$set", doc(pivotPath(pivotMigration) + ".column_groups.$[pivot].limit", pivotMigration.columnLimit.orElse(DEFAULT_LIMIT))),
                                    matchValuePivots
                            )
                    );
                    return builder.build().stream();
                })
                .collect(Collectors.toList());

        if (!operations.isEmpty()) {
            LOG.debug("Updating {} search types ...", pivotLimitMigrations.size());
            this.searches.bulkWrite(operations);
        }

        clusterConfigService.write(new MigrationCompleted(pivotLimitMigrations.size()));
    }
@Test
    @MongoDBFixtures("V20230113095301_MigrateGlobalPivotLimitsToGroupingsInSearchesTest_null_limits.json")
    void migratingPivotsWithNullLimits() {
        this.migration.upgrade();

        assertThat(migrationCompleted().migratedSearchTypes()).isEqualTo(2);

        final Document document = this.collection.find().first();

        final List<Document> searchTypes = getSearchTypes(document);

        for (Document searchType : searchTypes) {
            assertThatFieldsAreUnset(searchType);
        }
    }