Skip to content

Commit

Permalink
Persistence: purge unreferenced Objs
Browse files Browse the repository at this point in the history
This is an attempt to implement the algorithm mentioned in the PR #9401.

The `Obj.referenced()` attribute contains the timestamp when the object was last "referenced" (aka: attempted to be written). It is ...
* set when an object is first persisted via a `storeObj()`
* updated in the database, when an object was not persisted via `storeObj()`
* set/updated via `upsertObj()`
* updated via `updateConditional()`

Let's assume that there is a mechanism to identify the IDs of all referenced objects (it would be very similar to what the export functionality does). The algorithm to purge unreferenced objects must never delete an object that is referenced at any point of time, and must consider the case that an object that was unreferenced when a purge-unreferenced-objects routine started, but became referenced while it is running.

An approach could work as follows:

1. Memoize the current timestamp (minus some wall-clock drift adjustment).
2. Identify the IDs of all referenced objects. We could leverage a bloom filter, if the set of IDs is big.
3. Then scan all objects in the repository. Objects can be purged, if ...
    * the ID is not in the set (or bloom filter) generated in step 2 ...
    * _AND_ have a `referenced` timestamp less than the memoized timestamp.

Any deletion in the backing database would follow the meaning of this pseudo SQL: `DELETE FROM objs WHERE obj_id = :objId AND referenced < :memoizedTimestamp`.

Noting, that the `referenced` attribute is rather incorrect when retrieved from the objects cache (aka: during normal operations), which is not a problem, because that `referenced` attribute is irrelevant for production accesses.

There are two edge cases / race conditions:
* (for some backends): A `storeObj()` operation detected that the object already exists - then the purge routine deletes that object - and then the `storeObj()` tries to upddate the `referenced` attribute. The result is the loss of that object. This race condition can only occur, if the object existed but was not referenced.
* While the referenced objects are being identified, create a new named reference (branch / tag) pointing to commit(s) that would be identified as unreferenced and being later purged.
  • Loading branch information
snazy committed Oct 7, 2024
1 parent caac5a3 commit 2eec23c
Show file tree
Hide file tree
Showing 15 changed files with 758 additions and 0 deletions.
1 change: 1 addition & 0 deletions bom/build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@ dependencies {
api(project(":nessie-versioned-storage-dynamodb-tests"))
api(project(":nessie-versioned-storage-dynamodb2"))
api(project(":nessie-versioned-storage-dynamodb2-tests"))
api(project(":nessie-versioned-storage-garbage-collect"))
api(project(":nessie-versioned-storage-inmemory"))
api(project(":nessie-versioned-storage-inmemory-tests"))
api(project(":nessie-versioned-storage-jdbc"))
Expand Down
1 change: 1 addition & 0 deletions gradle/projects.main.properties
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ nessie-versioned-storage-dynamodb=versioned/storage/dynamodb
nessie-versioned-storage-dynamodb-tests=versioned/storage/dynamodb-tests
nessie-versioned-storage-dynamodb2=versioned/storage/dynamodb2
nessie-versioned-storage-dynamodb2-tests=versioned/storage/dynamodb2-tests
nessie-versioned-storage-garbage-collect=versioned/storage/garbage-collect
nessie-versioned-storage-inmemory=versioned/storage/inmemory
nessie-versioned-storage-inmemory-tests=versioned/storage/inmemory-tests
nessie-versioned-storage-jdbc=versioned/storage/jdbc
Expand Down
51 changes: 51 additions & 0 deletions versioned/storage/garbage-collect/build.gradle.kts
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* Copyright (C) 2022 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

plugins { id("nessie-conventions-server") }

publishingHelper { mavenName = "Nessie - Storage - Garbage Collect" }

description = "Identify and purge unreferenced objects."

dependencies {
implementation(project(":nessie-model"))
implementation(project(":nessie-versioned-storage-common"))
implementation(project(":nessie-versioned-spi"))
implementation(project(":nessie-versioned-transfer-related"))

compileOnly(libs.jakarta.validation.api)
compileOnly(libs.jakarta.annotation.api)
compileOnly(libs.microprofile.openapi)

compileOnly(platform(libs.jackson.bom))
compileOnly("com.fasterxml.jackson.core:jackson-annotations")

compileOnly(libs.errorprone.annotations)
implementation(libs.guava)

compileOnly(project(":nessie-versioned-storage-testextension"))

compileOnly(libs.immutables.builder)
compileOnly(libs.immutables.value.annotations)
annotationProcessor(libs.immutables.value.processor)

testImplementation(project(":nessie-versioned-storage-common-tests"))
testImplementation(project(":nessie-versioned-storage-inmemory"))
testImplementation(project(path = ":nessie-protobuf-relocated", configuration = "shadow"))
testImplementation(platform(libs.junit.bom))
testImplementation(libs.bundles.junit.testing)
testRuntimeOnly(libs.logback.classic)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

import jakarta.validation.constraints.NotNull;

public interface LiveObjectsResolver extends AutoCloseable {
void resolve(@NotNull ObjectsResolverContext objectsResolverContext)
throws MustRestartWithBiggerFilterException;
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

import static com.google.common.base.Preconditions.checkState;
import static org.projectnessie.versioned.storage.common.logic.CommitLogQuery.commitLogQuery;
import static org.projectnessie.versioned.storage.common.logic.Logics.commitLogic;
import static org.projectnessie.versioned.storage.common.logic.Logics.indexesLogic;
import static org.projectnessie.versioned.storage.common.logic.Logics.referenceLogic;
import static org.projectnessie.versioned.storage.common.logic.Logics.repositoryLogic;
import static org.projectnessie.versioned.storage.common.logic.ReferencesQuery.referencesQuery;

import org.projectnessie.model.Content;
import org.projectnessie.versioned.storage.common.exceptions.ObjNotFoundException;
import org.projectnessie.versioned.storage.common.indexes.StoreIndexElement;
import org.projectnessie.versioned.storage.common.objtypes.CommitObj;
import org.projectnessie.versioned.storage.common.objtypes.CommitOp;
import org.projectnessie.versioned.storage.common.objtypes.ContentValueObj;
import org.projectnessie.versioned.storage.common.objtypes.StandardObjType;
import org.projectnessie.versioned.storage.common.persist.Obj;
import org.projectnessie.versioned.storage.common.persist.ObjId;
import org.projectnessie.versioned.storage.common.persist.Reference;
import org.projectnessie.versioned.store.DefaultStoreWorker;

final class LiveObjectsResolverImpl implements LiveObjectsResolver {
@Override
public void resolve(ObjectsResolverContext objectsResolverContext)
throws MustRestartWithBiggerFilterException {
var persist = objectsResolverContext.persist();
checkState(
repositoryLogic(persist).repositoryExists(),
"The provided repository has not been initialized.");

objectsResolverContext
.relatedObjects()
.repositoryRelatedObjects()
.forEach(id -> handleObj(id, objectsResolverContext));

var referenceLogic = referenceLogic(persist);
referenceLogic
.queryReferences(referencesQuery())
.forEachRemaining(reference -> walkReference(reference, objectsResolverContext));
}

private void walkReference(Reference reference, ObjectsResolverContext objectsResolverContext) {
objectsResolverContext
.relatedObjects()
.referenceRelatedObjects(reference)
.forEach(id -> handleObj(id, objectsResolverContext));

commitLogic(objectsResolverContext.persist())
.commitLog(commitLogQuery(reference.pointer()))
.forEachRemaining(commit -> walkCommit(commit, objectsResolverContext));

var extendedInfo = reference.extendedInfoObj();
if (extendedInfo != null) {
objectsResolverContext.referencedObjects().markReferenced(extendedInfo);
}
}

private void walkCommit(CommitObj commit, ObjectsResolverContext objectsResolverContext) {
objectsResolverContext.referencedObjects().markReferenced(commit.id());

objectsResolverContext
.relatedObjects()
.commitRelatedObjects(commit)
.forEach(id -> handleObj(id, objectsResolverContext));

var indexesLogic = indexesLogic(objectsResolverContext.persist());
var index = indexesLogic.buildCompleteIndexOrEmpty(commit);
for (StoreIndexElement<CommitOp> indexElement : index) {
var content = indexElement.content();
if (content.action().exists()) {
var value = content.value();
handleObj(value, objectsResolverContext);
}
}
}

private void handleObj(ObjId value, ObjectsResolverContext objectsResolverContext) {
// TODO fetch 'value's in batches

Obj obj;
try {
obj = objectsResolverContext.persist().fetchObj(value);
} catch (ObjNotFoundException e) {
// ignore not-found situation - nothing what we could do here
return;
}

objectsResolverContext.referencedObjects().markReferenced(value);

var type = obj.type();

if (StandardObjType.VALUE.equals(type)) {
var contentValueObj = (ContentValueObj) obj;
var content =
DefaultStoreWorker.instance()
.valueFromStore(contentValueObj.payload(), contentValueObj.data());

handleContent(content, objectsResolverContext);
}
}

private void handleContent(Content content, ObjectsResolverContext objectsResolverContext) {
objectsResolverContext
.relatedObjects()
.contentRelatedObjects(content)
.forEach(id -> handleObj(id, objectsResolverContext));
}

@Override
public void close() {
// TODO handle remaining batched 'value's from handleValue().
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

/**
* Thrown when the bloom filter's FPP is above the configured threshold when adding IDs. If this
* exception is encountered, the current garbage-collection run <em>must</em> be aborted and
* restarted with a bigger {@link ObjectsResolverParams#expectedObjects()} value.
*/
public class MustRestartWithBiggerFilterException extends RuntimeException {
public MustRestartWithBiggerFilterException(String msg) {
super(msg);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

import jakarta.validation.constraints.NotNull;
import org.projectnessie.versioned.storage.common.persist.Persist;
import org.projectnessie.versioned.transfer.related.TransferRelatedObjects;

public interface ObjectsResolverContext {
@NotNull
Persist persist();

@NotNull
ReferencedObjects referencedObjects();

@NotNull
TransferRelatedObjects relatedObjects();

@NotNull
PurgeFilter purgeFilter();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

import org.projectnessie.versioned.storage.common.persist.Persist;
import org.projectnessie.versioned.transfer.related.TransferRelatedObjects;

final class ObjectsResolverContextImpl implements ObjectsResolverContext {
private final Persist persist;
private final ReferencedObjects filter;
private final TransferRelatedObjects relatedObjects;
private final PurgeFilter purgeFilter;

ObjectsResolverContextImpl(
Persist persist, ObjectsResolverParams params, PurgeFilter purgeFilter) {
this.persist = persist;
this.filter = new ReferencedObjectsImpl(params);
this.relatedObjects = params.relatedObjects();
this.purgeFilter = purgeFilter;
}

@Override
public Persist persist() {
return persist;
}

@Override
public ReferencedObjects referencedObjects() {
return filter;
}

@Override
public TransferRelatedObjects relatedObjects() {
return relatedObjects;
}

@Override
public PurgeFilter purgeFilter() {
return purgeFilter;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* Copyright (C) 2024 Dremio
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.projectnessie.versioned.storage.garbagecollect;

import org.immutables.value.Value;
import org.projectnessie.versioned.transfer.related.TransferRelatedObjects;

@Value.Immutable
public interface ObjectsResolverParams {
// Following defaults result in a serialized bloom filter size of about 3000000 bytes.
long DEFAULT_EXPECTED_FILE_COUNT = 1_000_000L;
double DEFAULT_FALSE_POSITIVE_PROBABILITY = 0.00001d;
double DEFAULT_ALLOWED_FALSE_POSITIVE_PROBABILITY = 0.0001d;

static ImmutableObjectsResolverParams.Builder builder() {
return ImmutableObjectsResolverParams.builder();
}

@Value.Default
default long expectedObjects() {
return DEFAULT_EXPECTED_FILE_COUNT;
}

@Value.Default
default double falsePositiveProbability() {
return DEFAULT_FALSE_POSITIVE_PROBABILITY;
}

@Value.Default
default double allowedFalsePositiveProbability() {
return DEFAULT_ALLOWED_FALSE_POSITIVE_PROBABILITY;
}

@Value.Default
default TransferRelatedObjects relatedObjects() {
return new TransferRelatedObjects() {};
}
}
Loading

0 comments on commit 2eec23c

Please sign in to comment.