NoSQL: Add maintenance implementation

bom/build.gradle.kts

@@ -67,6 +67,7 @@ dependencies {
     api(project(":polaris-persistence-nosql-mongodb"))
 
     api(project(":polaris-persistence-nosql-maintenance-api"))
+    api(project(":polaris-persistence-nosql-maintenance-impl"))
     api(project(":polaris-persistence-nosql-maintenance-cel"))
     api(project(":polaris-persistence-nosql-maintenance-spi"))
 

gradle/projects.main.properties

@@ -80,6 +80,7 @@ polaris-persistence-nosql-testextension=persistence/nosql/persistence/testextens
 polaris-persistence-nosql-varint=persistence/nosql/persistence/varint
 # persistence / maintenance
 polaris-persistence-nosql-maintenance-api=persistence/nosql/persistence/maintenance/api
+polaris-persistence-nosql-maintenance-impl=persistence/nosql/persistence/maintenance/impl
 polaris-persistence-nosql-maintenance-cel=persistence/nosql/persistence/maintenance/retain-cel
 polaris-persistence-nosql-maintenance-spi=persistence/nosql/persistence/maintenance/spi
 # persistence / database specific implementations

persistence/nosql/persistence/maintenance/impl/build.gradle.kts

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+plugins {
+  id("org.kordamp.gradle.jandex")
+  id("polaris-server")
+}
+
+description = "Polaris NoSQL persistence maintenance - service implementation"
+
+dependencies {
+  implementation(project(":polaris-persistence-nosql-api"))
+  implementation(project(":polaris-persistence-nosql-maintenance-api"))
+  implementation(project(":polaris-persistence-nosql-maintenance-spi"))
+  implementation(project(":polaris-persistence-nosql-realms-api"))
+  implementation(project(":polaris-idgen-api"))
+  runtimeOnly(project(":polaris-persistence-nosql-realms-impl"))
+  runtimeOnly(project(":polaris-persistence-nosql-realms-store-nosql"))
+
+  implementation(libs.guava)
+  implementation(libs.slf4j.api)
+
+  compileOnly(platform(libs.jackson.bom))
+  compileOnly("com.fasterxml.jackson.core:jackson-annotations")
+  compileOnly("com.fasterxml.jackson.core:jackson-databind")
+
+  compileOnly(libs.jakarta.annotation.api)
+  compileOnly(libs.jakarta.validation.api)
+  compileOnly(libs.jakarta.inject.api)
+  compileOnly(libs.jakarta.enterprise.cdi.api)
+
+  compileOnly(project(":polaris-immutables"))
+  annotationProcessor(project(":polaris-immutables", configuration = "processor"))
+
+  testFixturesApi(project(":polaris-persistence-nosql-api"))
+  testFixturesApi(project(":polaris-persistence-nosql-maintenance-api"))
+  testFixturesApi(project(":polaris-persistence-nosql-maintenance-spi"))
+  testFixturesApi(project(":polaris-persistence-nosql-testextension"))
+
+  testFixturesCompileOnly(project(":polaris-immutables"))
+  testFixturesAnnotationProcessor(project(":polaris-immutables", configuration = "processor"))
+
+  testFixturesCompileOnly(platform(libs.jackson.bom))
+  testFixturesCompileOnly("com.fasterxml.jackson.core:jackson-annotations")
+  testFixturesCompileOnly("com.fasterxml.jackson.core:jackson-databind")
+
+  testFixturesImplementation(libs.jakarta.annotation.api)
+  testFixturesImplementation(libs.jakarta.validation.api)
+  testFixturesCompileOnly(libs.jakarta.enterprise.cdi.api)
+
+  testCompileOnly(platform(libs.jackson.bom))
+  testCompileOnly("com.fasterxml.jackson.core:jackson-annotations")
+  testCompileOnly("com.fasterxml.jackson.core:jackson-databind")
+
+  testRuntimeOnly(libs.logback.classic)
+
+  testImplementation(project(":polaris-idgen-mocks"))
+  testRuntimeOnly(testFixtures(project(":polaris-persistence-nosql-cdi-weld")))
+  testImplementation(libs.weld.se.core)
+  testImplementation(libs.weld.junit5)
+  testRuntimeOnly(libs.smallrye.jandex)
+
+  testRuntimeOnly(project(":polaris-persistence-nosql-realms-impl"))
+  testRuntimeOnly(project(":polaris-persistence-nosql-realms-store-nosql"))
+  testRuntimeOnly(project(":polaris-persistence-nosql-inmemory"))
+  testImplementation(testFixtures(project(":polaris-persistence-nosql-inmemory")))
+}

persistence/nosql/persistence/maintenance/impl/src/main/java/org/apache/polaris/persistence/nosql/maintenance/impl/AbstractScanItemStatsCollector.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.persistence.nosql.maintenance.impl;
+
+import jakarta.annotation.Nonnull;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.polaris.persistence.nosql.api.obj.ObjRef;
+import org.apache.polaris.persistence.nosql.maintenance.api.MaintenanceRunInformation.MaintenanceStats;
+
+abstract class AbstractScanItemStatsCollector<I> implements ScanItemCallback<I> {
+  final StatsHolder stats = new StatsHolder();
+
+  /** Collect maintenance-run stats for objects per realm. */
+  static final class ScanRefStatsCollector extends AbstractScanItemStatsCollector<String> {
+    final Map<String, StatsHolder> perRealm = new HashMap<>();
+
+    /** Handles the maintenance-run outcome for a reference in a realm. */
+    @Override
+    public void itemOutcome(
+        @Nonnull String realm, @Nonnull String ref, @Nonnull ScanItemOutcome outcome) {
+      stats.add(outcome);
+      perRealm.computeIfAbsent(realm, realmId -> new StatsHolder()).add(outcome);
+    }
+
+    /** Retrieve maintenance-run reference stats per realm. */
+    Map<String, ? extends MaintenanceStats> toRealmObjTypeStatsMap() {
+      return perRealm.entrySet().stream()
+          .collect(Collectors.toMap(Map.Entry::getKey, r -> r.getValue().toMaintenanceStats()));
+    }
+  }
+
+  /**
+   * Collect maintenance-run stats for objects per realm and {@linkplain
+   * org.apache.polaris.persistence.nosql.api.obj.ObjType object type}.
+   */
+  static final class ScanObjStatsCollector extends AbstractScanItemStatsCollector<ObjRef> {
+    final Map<String, Map<String, StatsHolder>> perRealmAndObjType = new HashMap<>();
+
+    /** Handles the maintenance-run outcome for an object in a realm. */
+    @Override
+    public void itemOutcome(
+        @Nonnull String realm, @Nonnull ObjRef id, @Nonnull ScanItemOutcome outcome) {
+      stats.add(outcome);
+      perRealmAndObjType
+          .computeIfAbsent(realm, realmId -> new HashMap<>())
+          .computeIfAbsent(id.type(), objType -> new StatsHolder())
+          .add(outcome);
+    }
+
+    /**
+     * Retrieve maintenance-run reference stats per realm and {@linkplain
+     * org.apache.polaris.persistence.nosql.api.obj.ObjType object type}.
+     */
+    Map<String, ? extends Map<String, MaintenanceStats>> toRealmObjTypeStatsMap() {
+      return perRealmAndObjType.entrySet().stream()
+          .collect(
+              Collectors.toMap(
+                  Map.Entry::getKey,
+                  r ->
+                      r.getValue().entrySet().stream()
+                          .collect(
+                              Collectors.toMap(
+                                  Map.Entry::getKey, ot -> ot.getValue().toMaintenanceStats()))));
+    }
+  }
+
+  /** Maintenance stats holder. */
+  static final class StatsHolder {
+    long scanned;
+    long newer;
+    long retained;
+    long purged;
+
+    /** Consume the outcome for a reference or object-type. */
+    void add(ScanItemOutcome outcome) {
+      scanned++;
+      switch (outcome) {
+        case REALM_PURGE, PURGED -> purged++;
+        case TOO_NEW_RETAINED -> newer++;
+        case RETAINED, UNHANDLED_RETAINED -> retained++;
+        default -> throw new IllegalStateException("Unknown outcome " + outcome);
+      }
+    }
+
+    /** Produce the serializable-stats container. */
+    MaintenanceStats toMaintenanceStats() {
+      return MaintenanceStats.builder()
+          .scanned(scanned)
+          .newer(newer)
+          .retained(retained)
+          .purged(purged)
+          .build();
+    }
+  }
+}

persistence/nosql/persistence/maintenance/impl/src/main/java/org/apache/polaris/persistence/nosql/maintenance/impl/AllRetained.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.persistence.nosql.maintenance.impl;
+
+import static java.util.Map.entry;
+
+import com.google.common.hash.BloomFilter;
+import com.google.common.hash.PrimitiveSink;
+import jakarta.annotation.Nonnull;
+import java.util.Map;
+import java.util.function.Predicate;
+import org.apache.polaris.persistence.nosql.api.obj.ObjRef;
+import org.apache.polaris.persistence.nosql.maintenance.impl.ScanHandler.RetainCheck;
+import org.jspecify.annotations.NonNull;
+
+/**
+ * Collects reference names and objects to retain.
+ *
+ * <p>The implementation uses bloom-filters to limit the heap usage when a huge number of
+ * references/objects is being used.
+ */
+@SuppressWarnings("UnstableApiUsage")
+final class AllRetained {
+
+  /**
+   * Some "salt" to make the bloom filters non-deterministic, in case there are false-positives, to
+   * reduce the number of false positives over time.
+   */
+  private final int salt;
+
+  // @NonNull is the jspecify variant, which allows type-usage.
+  // Jakarta's @Nonnull does not allow type-usage.
+  private final BloomFilter<Map.@NonNull Entry<String, String>> refsFilter;
+  private final BloomFilter<Map.@NonNull Entry<String, Long>> objsFilter;
+  private long refAdds;
+  private long objAdds;
+
+  AllRetained(long expectedReferenceCount, long expectedObjCount, double fpp, int salt) {
+    this.salt = salt;
+    this.refsFilter = BloomFilter.create(this::refFunnel, expectedReferenceCount, fpp);
+    this.objsFilter = BloomFilter.create(this::objFunnel, expectedObjCount, fpp);
+  }
+
+  private void refFunnel(Map.Entry<String, String> realmRef, @Nonnull PrimitiveSink primitiveSink) {
+    primitiveSink.putInt(salt);
+    primitiveSink.putUnencodedChars(realmRef.getKey());
+    primitiveSink.putUnencodedChars(realmRef.getValue());
+  }
+
+  private void objFunnel(Map.Entry<String, Long> realmObj, @Nonnull PrimitiveSink primitiveSink) {
+    primitiveSink.putInt(salt);
+    primitiveSink.putUnencodedChars(realmObj.getKey());
+    var id = realmObj.getValue();
+    primitiveSink.putLong(id);
+  }
+
+  void addRetainedRef(String realm, String ref) {
+    refsFilter.put(entry(realm, ref));
+    refAdds++;
+  }
+
+  void addRetainedObj(String realm, long id) {
+    objsFilter.put(entry(realm, id));
+    objAdds++;
+  }
+
+  /** The number of {@link #addRetainedRef(String, String)} invocations. */
+  long refAdds() {
+    return refAdds;
+  }
+
+  /** The number of {@link #addRetainedObj(String, long)} invocations. */
+  long objAdds() {
+    return objAdds;
+  }
+
+  boolean withinExpectedFpp(double expectedFpp) {
+    return refsFilter.expectedFpp() < expectedFpp && objsFilter.expectedFpp() < expectedFpp;
+  }
+
+  RetainCheck<String> referenceRetainCheck() {
+    return (realm, ref) -> refsFilter.mightContain(entry(realm, ref));
+  }
+
+  RetainCheck<ObjRef> objRetainCheck(Predicate<String> objTypeIdPredicate) {
+    return (realm, id) ->
+        objTypeIdPredicate.test(id.type()) || objsFilter.mightContain(entry(realm, id.id()));
+  }
+}

persistence/nosql/persistence/maintenance/impl/src/main/java/org/apache/polaris/persistence/nosql/maintenance/impl/MaintenanceRunObj.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.persistence.nosql.maintenance.impl;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import java.util.function.LongSupplier;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.persistence.nosql.api.obj.AbstractObjType;
+import org.apache.polaris.persistence.nosql.api.obj.Obj;
+import org.apache.polaris.persistence.nosql.api.obj.ObjType;
+import org.apache.polaris.persistence.nosql.maintenance.api.MaintenanceRunInformation;
+
+/**
+ * Holds information about one maintenance run.
+ *
+ * <p>This object is <em>eventually overwritten</em> with the final result of the maintenance run.
+ */
+@PolarisImmutable
+@JsonSerialize(as = ImmutableMaintenanceRunObj.class)
+@JsonDeserialize(as = ImmutableMaintenanceRunObj.class)
+public interface MaintenanceRunObj extends Obj {
+
+  ObjType TYPE = new MaintenanceRunObjType();
+
+  @Override
+  default ObjType type() {
+    return TYPE;
+  }
+
+  MaintenanceRunInformation runInformation();
+
+  static ImmutableMaintenanceRunObj.Builder builder() {
+    return ImmutableMaintenanceRunObj.builder();
+  }
+
+  final class MaintenanceRunObjType extends AbstractObjType<MaintenanceRunObj> {
+    public MaintenanceRunObjType() {
+      super("mtr", "Maintenance Run", MaintenanceRunObj.class);
+    }
+
+    @Override
+    public long cachedObjectExpiresAtMicros(Obj obj, LongSupplier clockMicros) {
+      var mo = (MaintenanceRunObj) obj;
+      if (mo.runInformation().finished().isPresent()) {
+        return CACHE_UNLIMITED;
+      }
+      return NOT_CACHED;
+    }
+  }
+}