Skip to content

Commit

Permalink
Ensure that content IDs are unique in a Nessie repository (#7757)
Browse files Browse the repository at this point in the history
Nessie content IDs are random IDs, but we do not guarantee that those are actually really unique.

This change adds a new object type to ensure that a generated ID is unique by leveraging existing functionality of the `Persist` framework that already provides "`INSERT IF NOT EXIST`" guarantees.

New content IDs from this change on are now verified. This change does not include functionality to automatically add already existing content-IDs. IMHO it is probably okay for now given the practically non-existing probability of content-ID conflicts.
  • Loading branch information
snazy committed Dec 4, 2023
1 parent feeb0b7 commit b921c82
Show file tree
Hide file tree
Showing 18 changed files with 452 additions and 7 deletions.
16 changes: 13 additions & 3 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,16 +12,26 @@ as necessary. Empty sections will not end in the release notes.

- Nessie Quarkus parts are now built against Java 17 and Java 17 is required to run Nessie Quarkus Server directly.
If you use the Docker image, nothing needs to be done, because the image already contains a compatible Java runtime.
- Due to the introduction of extensible object types in the storage layer, some storage backends
- Due to the introduction of new object types in the storage layer, some storage backends
will require a schema upgrade:
- JDBC: the following SQL statement must be executed on the Nessie database (please adapt the
statement to the actual database SQL dialect):
```sql
ALTER TABLE objs ADD COLUMN x_class VARCHAR, ADD COLUMN x_data BYTEA, ADD COLUMN x_compress VARCHAR;
ALTER TABLE objs
ADD COLUMN x_class VARCHAR,
ADD COLUMN x_data BYTEA,
ADD COLUMN x_compress VARCHAR,
ADD COLUMN u_space VARCHAR,
ADD COLUMN u_value BYTEA;
```
- Cassandra: the following CQL statement must be executed on the Nessie database and keyspace:
```cql
ALTER TABLE <keyspace>.objs ADD x_class text, ADD x_data blob, ADD x_compress text;
ALTER TABLE <keyspace>.objs
ADD x_class text,
ADD x_data blob,
ADD x_compress text,
ADD u_space text,
ADD u_value blob;
```
- When using one of the legacy and deprecated version-store implementations based on "database adapter",
make sure to migrate to the new storage model **before** upgrading to this version or newer Nessie
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ private Persist base() {

private BatchingPersistImpl batching(Persist base) {
return (BatchingPersistImpl)
WriteBatching.builder().persist(base).batchSize(20).build().create();
WriteBatching.builder().persist(base).batchSize(21).build().create();
}

static Stream<Obj> allObjectTypeSamples() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ public final class ObjSerializers {
RefObjSerializer.INSTANCE,
StringObjSerializer.INSTANCE,
TagObjSerializer.INSTANCE,
UniqueIdObjSerializer.INSTANCE,
CustomObjSerializer.INSTANCE);

static {
Expand Down Expand Up @@ -75,6 +76,9 @@ public static ObjSerializer<Obj> forType(@Nonnull @jakarta.annotation.Nonnull Ob
case VALUE:
serializer = ContentValueObjSerializer.INSTANCE;
break;
case UNIQUE:
serializer = UniqueIdObjSerializer.INSTANCE;
break;
default:
throw new IllegalArgumentException("Unknown standard object type: " + type);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
/*
* Copyright (C) 2023 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.cassandra.serializers;

import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.INSERT_OBJ_PREFIX;
import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.INSERT_OBJ_VALUES;
import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.STORE_OBJ_SUFFIX;
import static org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj.uniqueId;

import com.datastax.oss.driver.api.core.cql.BoundStatementBuilder;
import com.datastax.oss.driver.api.core.cql.Row;
import com.google.common.collect.ImmutableSet;
import java.util.Set;
import java.util.stream.Collectors;
import org.projectnessie.versioned.storage.cassandra.CassandraSerde;
import org.projectnessie.versioned.storage.cassandra.CqlColumn;
import org.projectnessie.versioned.storage.cassandra.CqlColumnType;
import org.projectnessie.versioned.storage.common.exceptions.ObjTooLargeException;
import org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj;
import org.projectnessie.versioned.storage.common.persist.ObjId;

public class UniqueIdObjSerializer implements ObjSerializer<UniqueIdObj> {

public static final ObjSerializer<UniqueIdObj> INSTANCE = new UniqueIdObjSerializer();

private static final CqlColumn COL_UNIQUE_SPACE = new CqlColumn("u_space", CqlColumnType.VARCHAR);
private static final CqlColumn COL_UNIQUE_VALUE =
new CqlColumn("u_value", CqlColumnType.VARBINARY);

private static final Set<CqlColumn> COLS = ImmutableSet.of(COL_UNIQUE_SPACE, COL_UNIQUE_VALUE);

private static final String INSERT_CQL =
INSERT_OBJ_PREFIX
+ COLS.stream().map(CqlColumn::name).collect(Collectors.joining(","))
+ INSERT_OBJ_VALUES
+ COLS.stream().map(c -> ":" + c.name()).collect(Collectors.joining(","))
+ ")";

private static final String STORE_CQL = INSERT_CQL + STORE_OBJ_SUFFIX;

private UniqueIdObjSerializer() {}

@Override
public Set<CqlColumn> columns() {
return COLS;
}

@Override
public String insertCql(boolean upsert) {
return upsert ? INSERT_CQL : STORE_CQL;
}

@SuppressWarnings("ResultOfMethodCallIgnored")
@Override
public void serialize(
UniqueIdObj obj,
BoundStatementBuilder stmt,
int incrementalIndexLimit,
int maxSerializedIndexSize)
throws ObjTooLargeException {
stmt.setString(COL_UNIQUE_SPACE.name(), obj.space());
stmt.setByteBuffer(COL_UNIQUE_VALUE.name(), obj.value().asReadOnlyByteBuffer());
}

@Override
public UniqueIdObj deserialize(Row row, ObjId id) {
return uniqueId(
id,
row.getString(COL_UNIQUE_SPACE.name()),
CassandraSerde.deserializeBytes(row, COL_UNIQUE_VALUE.name()));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,11 @@ message StringProto {
bytes text = 54;
}

message UniqueIdProto {
string space = 1;
bytes value = 2;
}

enum CommitTypeProto {
T_UNKNOWN = 0;
NORMAL = 1;
Expand Down Expand Up @@ -108,6 +113,7 @@ message ObjProto {
StringProto string_data = 7;
TagProto tag = 8;
CustomProto custom = 9;
UniqueIdProto uniqueId = 10;
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import static org.projectnessie.versioned.storage.common.objtypes.RefObj.ref;
import static org.projectnessie.versioned.storage.common.objtypes.StringObj.stringData;
import static org.projectnessie.versioned.storage.common.objtypes.TagObj.tag;
import static org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj.uniqueId;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
Expand All @@ -49,6 +50,7 @@
import org.projectnessie.versioned.storage.common.objtypes.StandardObjType;
import org.projectnessie.versioned.storage.common.objtypes.StringObj;
import org.projectnessie.versioned.storage.common.objtypes.TagObj;
import org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj;
import org.projectnessie.versioned.storage.common.persist.ImmutableReference;
import org.projectnessie.versioned.storage.common.persist.Obj;
import org.projectnessie.versioned.storage.common.persist.ObjId;
Expand All @@ -68,6 +70,7 @@
import org.projectnessie.versioned.storage.common.proto.StorageTypes.StringProto;
import org.projectnessie.versioned.storage.common.proto.StorageTypes.Stripe;
import org.projectnessie.versioned.storage.common.proto.StorageTypes.TagProto;
import org.projectnessie.versioned.storage.common.proto.StorageTypes.UniqueIdProto;

public final class ProtoSerialization {

Expand Down Expand Up @@ -230,6 +233,8 @@ public static byte[] serializeObj(Obj obj, int incrementalIndexSizeLimit, int in
return b.setStringData(serializeStringData((StringObj) obj)).build().toByteArray();
case TAG:
return b.setTag(serializeTag((TagObj) obj)).build().toByteArray();
case UNIQUE:
return b.setUniqueId(serializeUniqueId((UniqueIdObj) obj)).build().toByteArray();
default:
throw new UnsupportedOperationException("Unknown standard object type " + obj.type());
}
Expand Down Expand Up @@ -284,6 +289,9 @@ public static Obj deserializeObjProto(ObjId id, ObjProto obj) {
if (obj.hasTag()) {
return deserializeTag(id, obj.getTag());
}
if (obj.hasUniqueId()) {
return deserializeUniqueId(id, obj.getUniqueId());
}
if (obj.hasCustom()) {
return deserializeCustom(id, obj.getCustom());
}
Expand Down Expand Up @@ -493,6 +501,14 @@ private static TagProto.Builder serializeTag(TagObj obj) {
return tag;
}

private static UniqueIdObj deserializeUniqueId(ObjId id, UniqueIdProto uniqueId) {
return uniqueId(id, uniqueId.getSpace(), uniqueId.getValue());
}

private static UniqueIdProto.Builder serializeUniqueId(UniqueIdObj obj) {
return UniqueIdProto.newBuilder().setSpace(obj.space()).setValue(obj.value());
}

private static Obj deserializeCustom(ObjId id, CustomProto custom) {
return SmileSerialization.deserializeObj(
id,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,12 @@
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.REF;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.STRING;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.TAG;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.UNIQUE;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.VALUE;
import static org.projectnessie.versioned.storage.common.objtypes.StringObj.stringData;
import static org.projectnessie.versioned.storage.common.objtypes.TagObj.tag;
import static org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj.uniqueId;
import static org.projectnessie.versioned.storage.common.objtypes.UniqueIdObj.uuidToBytes;
import static org.projectnessie.versioned.storage.common.persist.ObjId.EMPTY_OBJ_ID;
import static org.projectnessie.versioned.storage.common.persist.ObjId.objIdFromString;
import static org.projectnessie.versioned.storage.common.persist.ObjId.randomObjId;
Expand All @@ -70,6 +73,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.IntFunction;
import java.util.function.Supplier;
Expand Down Expand Up @@ -440,6 +444,7 @@ public static Stream<Obj> allObjectTypeSamples() {
copyFromUtf8("This is not a markdown")),
ref(randomObjId(), "foo", randomObjId(), 123L, null),
ref(randomObjId(), "bar", randomObjId(), 456L, randomObjId()),
uniqueId(randomObjId(), "space", uuidToBytes(UUID.randomUUID())),
// custom object types
SimpleCustomObj.builder()
.id(randomObjId())
Expand Down Expand Up @@ -481,6 +486,8 @@ static StandardObjType typeDifferentThan(ObjType type) {
return STRING;
case STRING:
return INDEX_SEGMENTS;
case UNIQUE:
return REF;
default:
// fall through
}
Expand Down Expand Up @@ -901,6 +908,8 @@ public static Obj updateObjChange(Obj obj) {
"filename",
asList(randomObjId(), randomObjId(), randomObjId(), randomObjId()),
ByteString.copyFrom(new byte[123]));
case UNIQUE:
return uniqueId(obj.id(), "other_space", uuidToBytes(UUID.randomUUID()));
default:
// fall through
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.REF;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.STRING;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.TAG;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.UNIQUE;
import static org.projectnessie.versioned.storage.common.objtypes.StandardObjType.VALUE;

import com.google.common.hash.Hasher;
Expand Down Expand Up @@ -122,4 +123,13 @@ static ObjId stringDataHash(
hasher.putBytes(text.asReadOnlyByteBuffer());
return hashAsObjId(hasher);
}

static ObjId uniqueIdHash(String space, ByteString value) {
Hasher hasher =
newHasher()
.putString(UNIQUE.name(), UTF_8)
.putString(space, UTF_8)
.putBytes(value.asReadOnlyByteBuffer());
return hashAsObjId(hasher);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,11 @@ public enum StandardObjType implements ObjType {
INDEX_SEGMENTS("I", IndexSegmentsObj.class),

/** {@link Obj} is a {@link IndexObj}. */
INDEX("i", IndexObj.class);
INDEX("i", IndexObj.class),

/** Obj is a {@link UniqueIdObj}. */
UNIQUE("u", UniqueIdObj.class),
;

private final String shortName;

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
/*
* 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.
*/
package org.projectnessie.versioned.storage.common.objtypes;

import static org.projectnessie.versioned.storage.common.objtypes.Hashes.uniqueIdHash;

import java.nio.ByteBuffer;
import java.util.UUID;
import javax.annotation.Nullable;
import org.immutables.value.Value;
import org.projectnessie.nessie.relocated.protobuf.ByteString;
import org.projectnessie.nessie.relocated.protobuf.UnsafeByteOperations;
import org.projectnessie.versioned.storage.common.persist.Obj;
import org.projectnessie.versioned.storage.common.persist.ObjId;
import org.projectnessie.versioned.storage.common.persist.ObjType;

/**
* Describes a unique ID, used to ensure that an ID defined in some {@linkplain #space()} is unique
* within a Nessie repository.
*/
@Value.Immutable
public interface UniqueIdObj extends Obj {

@Override
default ObjType type() {
return StandardObjType.UNIQUE;
}

@Override
@Value.Parameter(order = 1)
@Nullable
@jakarta.annotation.Nullable
ObjId id();

/** The "ID space", for example {@code content-id}. */
@Value.Parameter(order = 2)
String space();

/** The value of the ID within the {@link #space()}. */
@Value.Parameter(order = 3)
ByteString value();

@Value.NonAttribute
default UUID valueAsUUID() {
ByteBuffer buffer = value().asReadOnlyByteBuffer();
long msb = buffer.getLong();
long lsb = buffer.getLong();
return new UUID(msb, lsb);
}

static UniqueIdObj uniqueId(ObjId id, String space, ByteString value) {
return ImmutableUniqueIdObj.of(id, space, value);
}

static UniqueIdObj uniqueId(String space, ByteString value) {
return uniqueId(uniqueIdHash(space, value), space, value);
}

static UniqueIdObj uniqueId(String space, UUID value) {
return uniqueId(space, uuidToBytes(value));
}

static ByteString uuidToBytes(UUID value) {
ByteBuffer buffer = ByteBuffer.allocate(16);
buffer.putLong(value.getMostSignificantBits());
buffer.putLong(value.getLeastSignificantBits());
buffer.flip();
return UnsafeByteOperations.unsafeWrap(buffer);
}
}

0 comments on commit b921c82

Please sign in to comment.