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

        final List<ViewWidgetLimitMigration> widgetLimitMigrations = StreamSupport.stream(this.views.find().spliterator(), false)
                .flatMap(document -> {
                    final String viewId = document.get("_id", ObjectId.class).toHexString();
                    final Map<String, Document> state = document.get("state", Collections.emptyMap());
                    return state.entrySet().stream()
                            .flatMap(entry -> {
                                final String queryId = entry.getKey();
                                final List<Document> widgets = entry.getValue().get("widgets", Collections.emptyList());
                                return EntryStream.of(widgets)
                                        .filter(widget -> "aggregation".equals(widget.getValue().getString("type")))
                                        .flatMap(widgetEntry -> {
                                            final Document widget = widgetEntry.getValue();
                                            final Integer widgetIndex = widgetEntry.getKey();
                                            final Document config = widget.get("config", new Document());
                                            final boolean hasRowLimit = config.containsKey("row_limit");
                                            final boolean hasColumnLimit = config.containsKey("column_limit");
                                            final Optional<Integer> rowLimit = Optional.ofNullable(config.getInteger("row_limit"));
                                            final Optional<Integer> columnLimit = Optional.ofNullable(config.getInteger("column_limit"));

                                            if (widgetIndex != null && (hasRowLimit || hasColumnLimit)) {
                                                return Stream.of(new ViewWidgetLimitMigration(viewId, queryId, widgetIndex, rowLimit, columnLimit));
                                            }
                                            return Stream.empty();
                                        });
                            });
                })
                .collect(Collectors.toList());

        final List<WriteModel<Document>> operations = widgetLimitMigrations.stream()
                .flatMap(widgetMigration -> {
                    final ImmutableList.Builder<WriteModel<Document>> builder = ImmutableList.builder();
                    builder.add(
                            updateView(
                                    widgetMigration.viewId,
                                    doc("$unset", doc(widgetConfigPath(widgetMigration) + ".row_limit", 1))
                            )
                    );
                    builder.add(
                            updateView(
                                    widgetMigration.viewId,
                                    doc("$set", doc(widgetConfigPath(widgetMigration) + ".row_pivots.$[config].config.limit", widgetMigration.rowLimit.orElse(DEFAULT_LIMIT))),
                                    matchValuePivots
                            )
                    );
                    builder.add(
                            updateView(
                                    widgetMigration.viewId,
                                    doc("$unset", doc(widgetConfigPath(widgetMigration) + ".column_limit", 1))
                            )
                    );
                    builder.add(
                            updateView(
                                    widgetMigration.viewId,
                                    doc("$set", doc(widgetConfigPath(widgetMigration) + ".column_pivots.$[config].config.limit", widgetMigration.columnLimit.orElse(DEFAULT_LIMIT))),
                                    matchValuePivots
                            )
                    );
                    return builder.build().stream();
                })
                .collect(Collectors.toList());

        if (!operations.isEmpty()) {
            LOG.debug("Updating {} widgets ...", widgetLimitMigrations.size());
            this.views.bulkWrite(operations);
        }

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

        assertThat(migrationCompleted().migratedViews()).isEqualTo(1);

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

        final List<Document> widgets = getWidgets(document);

        for (Document widget : widgets) {
            assertThatFieldsAreUnset(widget);
        }
    }