Skip to content

Commit

Permalink
Refactor for polymorphism: object/list joiner
Browse files Browse the repository at this point in the history
  • Loading branch information
Sunjeet committed Aug 19, 2024
1 parent 70f3614 commit cd04995
Show file tree
Hide file tree
Showing 10 changed files with 142 additions and 99 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ public class HollowDiffUIServerTest {
public void test() throws Exception {
HollowDiff testDiff = new FakeHollowDiffGenerator().createFakeDiff();

HollowDiffUIServer server = new HollowDiffUIServer();
HollowDiffUIServer server = new HollowDiffUIServer(0);

server.addDiff("diff", testDiff);

Expand All @@ -22,7 +22,7 @@ public void test() throws Exception {
public void testBackwardsCompatibiltyWithJettyImplementation() throws Exception {
HollowDiff testDiff = new FakeHollowDiffGenerator().createFakeDiff();

com.netflix.hollow.diff.ui.jetty.HollowDiffUIServer server = new com.netflix.hollow.diff.ui.jetty.HollowDiffUIServer();
com.netflix.hollow.diff.ui.jetty.HollowDiffUIServer server = new com.netflix.hollow.diff.ui.jetty.HollowDiffUIServer(0);

server.addDiff("diff", testDiff);

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
package com.netflix.hollow.core.read.engine;

import com.netflix.hollow.core.memory.encoding.GapEncodedVariableLengthIntegerReader;

public abstract class AbstractHollowTypeDataElementsJoiner {
public final int fromMask;
public final int fromOrdinalShift;
public final AbstractHollowTypeDataElements[] from;

public AbstractHollowTypeDataElements to;

public AbstractHollowTypeDataElementsJoiner(AbstractHollowTypeDataElements[] from) {
this.from = from;
this.fromMask = from.length - 1;
this.fromOrdinalShift = 31 - Integer.numberOfLeadingZeros(from.length);

if (from.length<=0 || !((from.length&(from.length-1))==0)) {
throw new IllegalStateException("No. of DataElements to be joined must be a power of 2");
}

for (AbstractHollowTypeDataElements elements : from) {
if (elements.encodedAdditions != null) {
throw new IllegalStateException("Encountered encodedAdditions in data elements splitter- this is not expected " +
"since encodedAdditions only exist on delta data elements and they dont carry over to target data elements, " +
"delta data elements are never split/joined");
}
}
}

public AbstractHollowTypeDataElements join() {

init();

to.maxOrdinal = -1;

populateStats();

copyRecords();

GapEncodedVariableLengthIntegerReader[] fromRemovals = new GapEncodedVariableLengthIntegerReader[from.length];
for (int i=0;i<from.length;i++) {
fromRemovals[i] = from[i].encodedRemovals;
}
to.encodedRemovals = GapEncodedVariableLengthIntegerReader.join(fromRemovals);

return to;
}

public abstract void init();

public abstract void populateStats();

public abstract void copyRecords();


}
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,13 @@ public AbstractHollowTypeDataElementsSplitter(AbstractHollowTypeDataElements fro
}

public AbstractHollowTypeDataElements[] split() {

init();

for(int i=0;i<to.length;i++) {
to[i].maxOrdinal = -1;
}

populateStats();

copyRecords();
Expand All @@ -45,6 +49,8 @@ public AbstractHollowTypeDataElements[] split() {
return to;
}

public abstract void init();

public abstract void populateStats();

public abstract void copyRecords();
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
package com.netflix.hollow.core.read.engine.list;

import com.netflix.hollow.core.memory.FixedLengthDataFactory;
import com.netflix.hollow.core.memory.encoding.GapEncodedVariableLengthIntegerReader;
import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsJoiner;


/**
Expand All @@ -10,41 +10,22 @@
* The original data elements are not destroyed.
* The no. of passed data elements must be a power of 2.
*/
class HollowListTypeDataElementsJoiner {
class HollowListTypeDataElementsJoiner extends AbstractHollowTypeDataElementsJoiner {

HollowListTypeDataElements join(HollowListTypeDataElements[] from) {
final int fromMask = from.length - 1;
final int fromOrdinalShift = 31 - Integer.numberOfLeadingZeros(from.length);

if (from.length<=0 || !((from.length&(from.length-1))==0)) {
throw new IllegalStateException("No. of DataElements to be joined must be a power of 2");
}

HollowListTypeDataElements to = new HollowListTypeDataElements(from[0].memoryMode, from[0].memoryRecycler);

GapEncodedVariableLengthIntegerReader[] fromRemovals = new GapEncodedVariableLengthIntegerReader[from.length];
for (int i=0;i<from.length;i++) {
fromRemovals[i] = from[i].encodedRemovals;
}
to.encodedRemovals = GapEncodedVariableLengthIntegerReader.join(fromRemovals);

for (HollowListTypeDataElements elements : from) {
if (elements.encodedAdditions != null) {
throw new IllegalStateException("Encountered encodedAdditions in data elements joiner- this is not expected " +
"since encodedAdditions only exist on delta data elements and they dont carry over to target data elements, " +
"delta data elements are never split/joined");
}
}

populateStats(to, from);

copyRecords(to, from);
public HollowListTypeDataElementsJoiner(HollowListTypeDataElements[] from) {
super(from);
}

return to;
@Override
public void init() {
this.to = new HollowListTypeDataElements(from[0].memoryMode, from[0].memoryRecycler);
}

void populateStats(HollowListTypeDataElements to, HollowListTypeDataElements[] from) {
to.maxOrdinal = -1;
@Override
public void populateStats() {
HollowListTypeDataElements[] from = (HollowListTypeDataElements[]) this.from;
HollowListTypeDataElements to = (HollowListTypeDataElements) this.to;

for(int fromIndex=0;fromIndex<from.length;fromIndex++) {
int mappedMaxOrdinal = from[fromIndex].maxOrdinal == -1 ? -1 : (from[fromIndex].maxOrdinal * from.length) + fromIndex;
to.maxOrdinal = Math.max(to.maxOrdinal, mappedMaxOrdinal);
Expand All @@ -54,8 +35,6 @@ void populateStats(HollowListTypeDataElements to, HollowListTypeDataElements[] f
}
}

final int fromMask = from.length - 1;
final int fromOrdinalShift = 31 - Integer.numberOfLeadingZeros(from.length);
long totalOfListSizes = 0;
for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
int fromIndex = ordinal & fromMask;
Expand All @@ -82,9 +61,11 @@ void populateStats(HollowListTypeDataElements to, HollowListTypeDataElements[] f
to.totalNumberOfElements = totalOfListSizes;
}

private void copyRecords(HollowListTypeDataElements to, HollowListTypeDataElements[] from) {
final int fromMask = from.length - 1;
final int fromOrdinalShift = 31 - Integer.numberOfLeadingZeros(from.length);
@Override
public void copyRecords() {
HollowListTypeDataElements[] from = (HollowListTypeDataElements[]) this.from;
HollowListTypeDataElements to = (HollowListTypeDataElements) this.to;

long elementCounter = 0;
for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
int fromIndex = ordinal & fromMask;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ public class HollowListTypeDataElementsSplitter extends AbstractHollowTypeDataEl

public HollowListTypeDataElementsSplitter(HollowListTypeDataElements from, int numSplits) {
super(from, numSplits);
}

@Override
public void init() {
this.to = new HollowListTypeDataElements[numSplits];
for(int i=0;i<to.length;i++) {
to[i] = new HollowListTypeDataElements(from.memoryMode, from.memoryRecycler);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@

import com.netflix.hollow.core.memory.FixedLengthDataFactory;
import com.netflix.hollow.core.memory.VariableLengthDataFactory;
import com.netflix.hollow.core.memory.encoding.GapEncodedVariableLengthIntegerReader;
import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsJoiner;
import com.netflix.hollow.core.schema.HollowObjectSchema;


/**
Expand All @@ -15,62 +16,27 @@
* The original data elements are not destroyed.
* The no. of passed data elements must be a power of 2.
*/
class HollowObjectTypeDataElementsJoiner {
class HollowObjectTypeDataElementsJoiner extends AbstractHollowTypeDataElementsJoiner {

HollowObjectTypeDataElements join(HollowObjectTypeDataElements[] from) {
final int fromMask = from.length - 1;
final int fromOrdinalShift = 31 - Integer.numberOfLeadingZeros(from.length);
long[] currentWriteVarLengthDataPointers;
private HollowObjectSchema schema;
private HollowObjectTypeDataElements to;

if (from.length<=0 || !((from.length&(from.length-1))==0)) {
throw new IllegalStateException("No. of DataElements to be joined must be a power of 2");
}

HollowObjectTypeDataElements to = new HollowObjectTypeDataElements(from[0].schema, from[0].memoryMode, from[0].memoryRecycler);
currentWriteVarLengthDataPointers = new long[from[0].schema.numFields()];

populateStats(to, from);

GapEncodedVariableLengthIntegerReader[] fromRemovals = new GapEncodedVariableLengthIntegerReader[from.length];
for (int i=0;i<from.length;i++) {
fromRemovals[i] = from[i].encodedRemovals;
}
to.encodedRemovals = GapEncodedVariableLengthIntegerReader.join(fromRemovals);

for (HollowObjectTypeDataElements elements : from) {
if (elements.encodedAdditions != null) {
throw new IllegalStateException("Encountered encodedAdditions in data elements joiner- this is not expected " +
"since encodedAdditions only exist on delta data elements and they dont carry over to target data elements, " +
"delta data elements are never split/joined");
}
}

for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
int fromIndex = ordinal & fromMask;
int fromOrdinal = ordinal >> fromOrdinalShift;

if (fromOrdinal <= from[fromIndex].maxOrdinal) {
copyRecord(to, ordinal, from[fromIndex], fromOrdinal, currentWriteVarLengthDataPointers);
} else {
// lopsided shards could result for consumers that skip type shards with no additions
writeNullRecord(to, ordinal, currentWriteVarLengthDataPointers);
}
}

return to;
public HollowObjectTypeDataElementsJoiner(HollowObjectTypeDataElements[] from) {
super(from);
this.schema = from[0].schema;
}

private void writeNullRecord(HollowObjectTypeDataElements to, int toOrdinal, long[] currentWriteVarLengthDataPointers) {
for(int fieldIndex=0;fieldIndex<to.schema.numFields();fieldIndex++) {
long currentWriteFixedLengthStartBit = ((long)toOrdinal * to.bitsPerRecord) + to.bitOffsetPerField[fieldIndex];
writeNullField(to, fieldIndex, currentWriteFixedLengthStartBit, currentWriteVarLengthDataPointers);
}
@Override
public void init() {
this.to = new HollowObjectTypeDataElements(schema, from[0].memoryMode, from[0].memoryRecycler);
}

void populateStats(HollowObjectTypeDataElements to, HollowObjectTypeDataElements[] from) {
@Override
public void populateStats() {
HollowObjectTypeDataElements[] from = (HollowObjectTypeDataElements[]) this.from;

long[] varLengthSizes = new long[to.schema.numFields()];

to.maxOrdinal = -1;
for(int fromIndex=0;fromIndex<from.length;fromIndex++) {
for(int ordinal=0;ordinal<=from[fromIndex].maxOrdinal;ordinal++) {
for(int fieldIdx=0;fieldIdx<to.schema.numFields();fieldIdx++) {
Expand Down Expand Up @@ -111,4 +77,31 @@ void populateStats(HollowObjectTypeDataElements to, HollowObjectTypeDataElements
// to.bitsPerUnfilteredField
// to.unfilteredFieldIsIncluded
}

@Override
public void copyRecords() {
HollowObjectTypeDataElements[] from = (HollowObjectTypeDataElements[]) this.from;

long[] currentWriteVarLengthDataPointers = new long[from[0].schema.numFields()];

for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
int fromIndex = ordinal & fromMask;
int fromOrdinal = ordinal >> fromOrdinalShift;

if (fromOrdinal <= from[fromIndex].maxOrdinal) {
copyRecord(to, ordinal, from[fromIndex], fromOrdinal, currentWriteVarLengthDataPointers);
} else {
// lopsided shards could result for consumers that skip type shards with no additions
writeNullRecord(to, ordinal, currentWriteVarLengthDataPointers);
}
}

}

private void writeNullRecord(HollowObjectTypeDataElements to, int toOrdinal, long[] currentWriteVarLengthDataPointers) {
for(int fieldIndex=0;fieldIndex<to.schema.numFields();fieldIndex++) {
long currentWriteFixedLengthStartBit = ((long)toOrdinal * to.bitsPerRecord) + to.bitOffsetPerField[fieldIndex];
writeNullField(to, fieldIndex, currentWriteFixedLengthStartBit, currentWriteVarLengthDataPointers);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -7,21 +7,27 @@
import com.netflix.hollow.core.memory.VariableLengthDataFactory;
import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElements;
import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsSplitter;
import com.netflix.hollow.core.schema.HollowObjectSchema;

/**
* Split a {@code HollowObjectTypeDataElements} into multiple {@code HollowObjectTypeDataElements}s.
* Ordinals are remapped and corresponding data is copied over.
* The original data elements are not destroyed.
* // SNAP: TODO: single-shot, split() not expected to be called repeatedly
* {@code numSplits} must be a power of 2.
*/
public class HollowObjectTypeDataElementsSplitter extends AbstractHollowTypeDataElementsSplitter {
private HollowObjectSchema schema;

HollowObjectTypeDataElementsSplitter(HollowObjectTypeDataElements from, int numSplits) {
super(from, numSplits);
this.schema = from.schema;
}

@Override
public void init() {
this.to = new HollowObjectTypeDataElements[numSplits];
for(int i=0;i<to.length;i++) {
to[i] = new HollowObjectTypeDataElements(from.schema, from.memoryMode, from.memoryRecycler);
to[i] = new HollowObjectTypeDataElements(schema, from.memoryMode, from.memoryRecycler);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@ public class HollowSetTypeDataElementsSplitter extends AbstractHollowTypeDataEle

public HollowSetTypeDataElementsSplitter(HollowSetTypeDataElements from, int numSplits) {
super(from, numSplits);
}

@Override
public void init() {
this.to = new HollowSetTypeDataElements[numSplits];
for(int i=0;i<to.length;i++) {
to[i] = new HollowSetTypeDataElements(from.memoryMode, from.memoryRecycler);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ public void calculateSnapshot() {

for(int i=0;i<numShards;i++) {
listPointerArray[i] = new FixedLengthElementArray(WastefulRecycler.DEFAULT_INSTANCE, (long)bitsPerListPointer * (maxShardOrdinal[i] + 1));
elementArray[i] = new FixedLengthElementArray(WastefulRecycler.DEFAULT_INSTANCE, (long)bitsPerElement * totalOfListSizes[i]); // SNAP: TODO: elementData
elementArray[i] = new FixedLengthElementArray(WastefulRecycler.DEFAULT_INSTANCE, (long)bitsPerElement * totalOfListSizes[i]);
}

ByteData data = ordinalMap.getByteData().getUnderlyingArray();
Expand Down Expand Up @@ -216,7 +216,7 @@ private void writeSnapshotShard(DataOutputStream os, int shardNumber) throws IOE
/// 2) statistics
VarInt.writeVInt(os, bitsPerListPointer);
VarInt.writeVInt(os, bitsPerElement);
VarInt.writeVLong(os, totalOfListSizes[shardNumber]); // SNAP: TODO: totalNumberOfElements
VarInt.writeVLong(os, totalOfListSizes[shardNumber]);

/// 3) list pointer array
int numListPointerLongs = maxShardOrdinal[shardNumber] == -1 ? 0 : (int)((((long)(maxShardOrdinal[shardNumber] + 1) * bitsPerListPointer) - 1) / 64) + 1;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,13 +74,6 @@ static void prettyPrintArray(int[][] array) {
}
}

// SNAP: TODO: remove
// static int fibonacci(int n) {
// if (n <= 1) {
// return n;
// }
// return fibonacci(n - 1) + fibonacci(n - 2);
// }
private void assertChecksumUnchanged(HollowListTypeReadState newTypeState, HollowListTypeReadState origTypeState, BitSet populatedOrdinals) {
HollowChecksum origCksum = new HollowChecksum();
HollowChecksum newCksum = new HollowChecksum();
Expand Down

0 comments on commit cd04995

Please sign in to comment.