Skip to content

Precompute PartitionFunction key to reduce broker pruning CPU #224

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
praveenc7 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Precompute PartitionFunction key to reduce broker pruning CPU #224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions ...main/java/org/apache/pinot/segment/spi/partition/BoundedColumnValuePartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public class BoundedColumnValuePartitionFunction implements PartitionFunction {
private final int _numPartitions;
private final Map<String, String> _functionConfig;
private final String[] _values;
private final String _partitionFunctionKey;

public BoundedColumnValuePartitionFunction(int numPartitions, Map<String, String> functionConfig) {
Preconditions.checkArgument(functionConfig != null && functionConfig.size() > 0,
Expand All @@ -59,6 +60,8 @@ public BoundedColumnValuePartitionFunction(int numPartitions, Map<String, String
Preconditions.checkState(numPartitions == _values.length + 1,
"'numPartitions' must just be one greater than number of column values configured");
_numPartitions = numPartitions;
_partitionFunctionKey = NAME + "_" + _numPartitions + "_" + _functionConfig.get(COLUMN_VALUES)
+ "_" + _functionConfig.get(COLUMN_VALUES_DELIMITER);
}

@Override
Expand Down Expand Up @@ -98,7 +101,6 @@ public String toString() {

@Override
public String getPartitionFunctionKey() {
return NAME + "_" + _numPartitions + "_" + _functionConfig.get(COLUMN_VALUES)
+ "_" + _functionConfig.get(COLUMN_VALUES_DELIMITER);
return _partitionFunctionKey;
}
}
7 changes: 7 additions & 0 deletions ...-spi/src/main/java/org/apache/pinot/segment/spi/partition/ByteArrayPartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
public class ByteArrayPartitionFunction implements PartitionFunction {
private static final String NAME = "ByteArray";
private final int _numPartitions;
private final String _partitionFunctionKey;

/**
* Constructor for the class.
Expand All @@ -39,6 +40,7 @@ public class ByteArrayPartitionFunction implements PartitionFunction {
public ByteArrayPartitionFunction(int numPartitions) {
Preconditions.checkArgument(numPartitions > 0, "Number of partitions must be > 0, specified", numPartitions);
_numPartitions = numPartitions;
_partitionFunctionKey = NAME + "_" + _numPartitions;
}

@Override
Expand All @@ -56,6 +58,11 @@ public int getNumPartitions() {
return _numPartitions;
}

@Override
public String getPartitionFunctionKey() {
return _partitionFunctionKey;
}

// Keep it for backward-compatibility, use getName() instead
@Override
public String toString() {
Expand Down
7 changes: 7 additions & 0 deletions ...t-spi/src/main/java/org/apache/pinot/segment/spi/partition/HashCodePartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,12 @@
public class HashCodePartitionFunction implements PartitionFunction {
private static final String NAME = "HashCode";
private final int _numPartitions;
private final String _partitionFunctionKey;

public HashCodePartitionFunction(int numPartitions) {
Preconditions.checkArgument(numPartitions > 0, "Number of partitions must be > 0, specified", numPartitions);
_numPartitions = numPartitions;
_partitionFunctionKey = NAME + "_" + _numPartitions;
}

@Override
Expand All @@ -51,6 +53,11 @@ public int getNumPartitions() {
return _numPartitions;
}

@Override
public String getPartitionFunctionKey() {
return _partitionFunctionKey;
}

// Keep it for backward-compatibility, use getName() instead
@Override
public String toString() {
Expand Down
7 changes: 7 additions & 0 deletions ...ent-spi/src/main/java/org/apache/pinot/segment/spi/partition/ModuloPartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
public class ModuloPartitionFunction implements PartitionFunction {
private static final String NAME = "Modulo";
private final int _numPartitions;
private final String _partitionFunctionKey;

/**
* Constructor for the class.
Expand All @@ -39,6 +40,7 @@ public class ModuloPartitionFunction implements PartitionFunction {
public ModuloPartitionFunction(int numPartitions) {
Preconditions.checkArgument(numPartitions > 0, "Number of partitions must be > 0, specified", numPartitions);
_numPartitions = numPartitions;
_partitionFunctionKey = NAME + "_" + _numPartitions;
}

/**
Expand All @@ -63,6 +65,11 @@ public int getNumPartitions() {
return _numPartitions;
}

@Override
public String getPartitionFunctionKey() {
return _partitionFunctionKey;
}

// Keep it for backward-compatibility, use getName() instead
@Override
public String toString() {
Expand Down
4 changes: 3 additions & 1 deletion ...nt-spi/src/main/java/org/apache/pinot/segment/spi/partition/Murmur3PartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class Murmur3PartitionFunction implements PartitionFunction {
private final int _numPartitions;
private final int _seed;
private final boolean _useX64;
private final String _partitionFunctionKey;

/**
* Constructor for the class.
Expand Down Expand Up @@ -65,6 +66,7 @@ public Murmur3PartitionFunction(int numPartitions, Map<String, String> functionC
}
_seed = seed;
_useX64 = useX64;
_partitionFunctionKey = NAME + "_" + _numPartitions + "_" + _seed + "_" + (_useX64 ? "x64" : "x86");
}

@Override
Expand Down Expand Up @@ -92,6 +94,6 @@ public String toString() {

@Override
public String getPartitionFunctionKey() {
return NAME + "_" + _numPartitions + "_" + _seed + "_" + (_useX64 ? "x64" : "x86");
return _partitionFunctionKey;
}
}
7 changes: 7 additions & 0 deletions ...ent-spi/src/main/java/org/apache/pinot/segment/spi/partition/MurmurPartitionFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
public class MurmurPartitionFunction implements PartitionFunction {
private static final String NAME = "Murmur";
private final int _numPartitions;
private final String _partitionFunctionKey;

/**
* Constructor for the class.
Expand All @@ -38,6 +39,7 @@ public class MurmurPartitionFunction implements PartitionFunction {
public MurmurPartitionFunction(int numPartitions) {
Preconditions.checkArgument(numPartitions > 0, "Number of partitions must be > 0");
_numPartitions = numPartitions;
_partitionFunctionKey = NAME + "_" + _numPartitions;
}

@Override
Expand All @@ -55,6 +57,11 @@ public int getNumPartitions() {
return _numPartitions;
}

@Override
public String getPartitionFunctionKey() {
return _partitionFunctionKey;
}

// Keep it for backward-compatibility, use getName() instead
@Override
public String toString() {
Expand Down
26 changes: 26 additions & 0 deletions ...gment-spi/src/test/java/org/apache/pinot/segment/spi/partition/PartitionFunctionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;


Expand Down Expand Up @@ -417,6 +418,31 @@ public void testPartitionFunctionKey() {
assertTrue(!murmur3Function.getPartitionFunctionKey().equals(murmur3DifferentSeed.getPartitionFunctionKey()));
}

@Test
public void testPartitionFunctionKeyIsPrecomputed() {
// The key is computed once and called per-segment on the broker's hot pruning path. Verify each implementation
// returns the same cached instance on repeated calls rather than rebuilding the string every time.
Map<String, String> murmur3Config = new HashMap<>();
murmur3Config.put("seed", "42");
murmur3Config.put("variant", "x64_32");
Map<String, String> boundedConfig = new HashMap<>();
boundedConfig.put("columnValues", "a|b|c");
boundedConfig.put("columnValuesDelimiter", "|");

PartitionFunction[] functions = {
PartitionFunctionFactory.getPartitionFunction("Modulo", 10, null),
PartitionFunctionFactory.getPartitionFunction("HashCode", 20, null),
PartitionFunctionFactory.getPartitionFunction("Murmur", 15, null),
PartitionFunctionFactory.getPartitionFunction("Murmur3", 25, murmur3Config),
PartitionFunctionFactory.getPartitionFunction("ByteArray", 12, null),
PartitionFunctionFactory.getPartitionFunction("BoundedColumnValue", 4, boundedConfig)
};
for (PartitionFunction function : functions) {
assertSame(function.getPartitionFunctionKey(), function.getPartitionFunctionKey(),
function.getName() + " should return the same precomputed key instance on every call");
}
}

private void testBasicProperties(PartitionFunction partitionFunction, String functionName, int numPartitions) {
testBasicProperties(partitionFunction, functionName, numPartitions, null);
}
Expand Down
Loading