wave10b/c: 465/212 hudi/iceberg/scylladb/yugabyte/foundationdb
This commit is contained in:
parent
f7fa333977
commit
70702dff5c
38 changed files with 2073 additions and 5 deletions
|
|
@ -0,0 +1,99 @@
|
|||
# doris-0001: BindExpression.processNonStandardAggregate — List.contains per projection → O(P×G)
|
||||
|
||||
## Classification
|
||||
- **Severity**: MEDIUM
|
||||
- **CWE**: CWE-407 (Algorithmic Complexity — Inefficient Algorithmic Complexity)
|
||||
- **Component**: `fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java`
|
||||
- **Method**: `processNonStandardAggregate()`
|
||||
|
||||
## Defect
|
||||
|
||||
`processNonStandardAggregate()` accepts `Collection<Expression> groupingExprs`. It is called from
|
||||
two sites:
|
||||
|
||||
- Line 1401: `processNonStandardAggregate(boundProjections, boundGroupBy)` where `boundGroupBy` is
|
||||
a `List<Expression>` (built by `ImmutableList.Builder` in `bindGroupBy()`).
|
||||
- Line 1578: `processNonStandardAggregate(boundOutput, flatBoundGroupingSet)` where
|
||||
`flatBoundGroupingSet` is a `Set<Expression>` — this call is safe.
|
||||
|
||||
For the first (List) call-site:
|
||||
|
||||
```java
|
||||
for (NamedExpression projection : originalProjections) { // P projections
|
||||
if (projection instanceof SlotReference && !groupingExprs.contains(projection)) { // O(G) List scan
|
||||
...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Each of P projections calls `groupingExprs.contains()` which is O(G) for a `List`. Total: **O(P × G)**.
|
||||
|
||||
In non-FULL_GROUP_BY SQL mode (MySQL-compatible mode), this code path is taken for every aggregate
|
||||
query. Wide SELECT lists with many GROUP BY keys degrade quadratically.
|
||||
|
||||
## Root Cause
|
||||
|
||||
`bindGroupBy()` returns `List<Expression>`. The result is passed directly to
|
||||
`processNonStandardAggregate()` without converting to a `Set`. The `Collection<>` parameter type
|
||||
masks the problem at the call site.
|
||||
|
||||
## Fix
|
||||
|
||||
Wrap `groupingExprs` in a `HashSet` at the start of `processNonStandardAggregate()` when it is not
|
||||
already a `Set`, or tighten the API to accept `Set<Expression>` and update call sites.
|
||||
|
||||
Option A (minimal, defensive):
|
||||
```java
|
||||
private List<NamedExpression> processNonStandardAggregate(
|
||||
List<NamedExpression> originalProjections, Collection<Expression> groupingExprs) {
|
||||
if (SqlModeHelper.hasOnlyFullGroupBy()) {
|
||||
return originalProjections;
|
||||
}
|
||||
// Ensure O(1) membership: convert to Set if caller passed a List
|
||||
Set<Expression> groupingSet = (groupingExprs instanceof Set)
|
||||
? (Set<Expression>) groupingExprs
|
||||
: new HashSet<>(groupingExprs);
|
||||
ImmutableList.Builder<NamedExpression> finalProjectionsBuilder = ImmutableList.builder();
|
||||
for (NamedExpression projection : originalProjections) {
|
||||
if (projection instanceof SlotReference && !groupingSet.contains(projection)) {
|
||||
finalProjectionsBuilder.add(new Alias(projection, projection.getName()));
|
||||
} else {
|
||||
finalProjectionsBuilder.add(projection);
|
||||
}
|
||||
}
|
||||
return finalProjectionsBuilder.build();
|
||||
}
|
||||
```
|
||||
|
||||
## Complexity
|
||||
|
||||
| Before | After |
|
||||
|--------|-------|
|
||||
| O(P × G) | O(P + G) |
|
||||
|
||||
With P=50 projections, G=40 group-by keys: Before = 2 000 comparisons. After = 90. **~22× reduction**.
|
||||
|
||||
## Patch
|
||||
|
||||
```diff
|
||||
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java
|
||||
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java
|
||||
@@ -1598,9 +1598,13 @@ public class BindExpression extends OneRewriteRuleFactory {
|
||||
private List<NamedExpression> processNonStandardAggregate(
|
||||
List<NamedExpression> originalProjections, Collection<Expression> groupingExprs) {
|
||||
if (SqlModeHelper.hasOnlyFullGroupBy()) {
|
||||
return originalProjections;
|
||||
} else {
|
||||
+ // Ensure O(1) membership — callers may pass List<Expression>
|
||||
+ Set<Expression> groupingSet = (groupingExprs instanceof Set)
|
||||
+ ? (Set<Expression>) groupingExprs
|
||||
+ : new HashSet<>(groupingExprs);
|
||||
ImmutableList.Builder<NamedExpression> finalProjectionsBuilder = ImmutableList.builder();
|
||||
for (NamedExpression projection : originalProjections) {
|
||||
// we do a trick here
|
||||
- if (projection instanceof SlotReference && !groupingExprs.contains(projection)) {
|
||||
+ if (projection instanceof SlotReference && !groupingSet.contains(projection)) {
|
||||
finalProjectionsBuilder.add(new Alias(projection, projection.getName()));
|
||||
} else {
|
||||
finalProjectionsBuilder.add(projection);
|
||||
```
|
||||
181
defects/doris/unit/BindExprGroupingAlgorithm.java
Normal file
181
defects/doris/unit/BindExprGroupingAlgorithm.java
Normal file
|
|
@ -0,0 +1,181 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.*;
|
||||
|
||||
/**
|
||||
* Standalone unit test for doris-0001:
|
||||
* BindExpression.processNonStandardAggregate — List.contains() per projection → O(P×G).
|
||||
*
|
||||
* Simulates processNonStandardAggregate():
|
||||
* for each projection, check if it's in groupingExprs (passed as List).
|
||||
*
|
||||
* Compile: javac -d . BindExprGroupingAlgorithm.java
|
||||
* Run: java unit.BindExprGroupingAlgorithm
|
||||
*/
|
||||
public class BindExprGroupingAlgorithm {
|
||||
|
||||
// ── Expression stubs ─────────────────────────────────────────────────────
|
||||
|
||||
static class Expression {
|
||||
final String id;
|
||||
Expression(String id) { this.id = id; }
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof Expression && id.equals(((Expression) o).id);
|
||||
}
|
||||
@Override public int hashCode() { return id.hashCode(); }
|
||||
@Override public String toString() { return id; }
|
||||
}
|
||||
|
||||
static class SlotReference extends Expression {
|
||||
SlotReference(String id) { super(id); }
|
||||
}
|
||||
|
||||
// ── Result ───────────────────────────────────────────────────────────────
|
||||
|
||||
static class Result {
|
||||
final List<String> output; // projection labels: "alias:X" or "keep:X"
|
||||
final long ns;
|
||||
Result(List<String> output, long ns) { this.output = output; this.ns = ns; }
|
||||
}
|
||||
|
||||
// ── Defective: Collection<Expression> as List → O(G) .contains per proj ─
|
||||
|
||||
static class DefectiveAlgorithm {
|
||||
Result processNonStandardAggregate(
|
||||
List<Expression> originalProjections,
|
||||
Collection<Expression> groupingExprs) { // List passed → O(G) .contains
|
||||
long t0 = System.nanoTime();
|
||||
List<String> out = new ArrayList<>();
|
||||
for (Expression projection : originalProjections) {
|
||||
if (projection instanceof SlotReference
|
||||
&& !groupingExprs.contains(projection)) { // CWE-407 site
|
||||
out.add("alias:" + projection.id);
|
||||
} else {
|
||||
out.add("keep:" + projection.id);
|
||||
}
|
||||
}
|
||||
return new Result(out, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Fixed: convert to Set once → O(1) .contains ──────────────────────────
|
||||
|
||||
static class FixedAlgorithm {
|
||||
Result processNonStandardAggregate(
|
||||
List<Expression> originalProjections,
|
||||
Collection<Expression> groupingExprs) {
|
||||
long t0 = System.nanoTime();
|
||||
// Ensure O(1) membership
|
||||
Set<Expression> groupingSet = (groupingExprs instanceof Set)
|
||||
? (Set<Expression>) groupingExprs
|
||||
: new HashSet<>(groupingExprs); // O(G) once
|
||||
List<String> out = new ArrayList<>();
|
||||
for (Expression projection : originalProjections) {
|
||||
if (projection instanceof SlotReference
|
||||
&& !groupingSet.contains(projection)) { // O(1)
|
||||
out.add("alias:" + projection.id);
|
||||
} else {
|
||||
out.add("keep:" + projection.id);
|
||||
}
|
||||
}
|
||||
return new Result(out, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
static List<Expression> makeProjections(int P, int G) {
|
||||
// First G projections are SlotReferences that ARE in groupingExprs → "keep"
|
||||
// Next (P-G) are SlotReferences NOT in groupingExprs → "alias"
|
||||
List<Expression> list = new ArrayList<>(P);
|
||||
for (int i = 0; i < P; i++) {
|
||||
list.add(new SlotReference("s" + i));
|
||||
}
|
||||
return list;
|
||||
}
|
||||
|
||||
static List<Expression> makeGroupingList(int G) {
|
||||
List<Expression> list = new ArrayList<>(G);
|
||||
for (int i = 0; i < G; i++) list.add(new SlotReference("s" + i));
|
||||
return list;
|
||||
}
|
||||
|
||||
// ── Assertions ───────────────────────────────────────────────────────────
|
||||
|
||||
static void assertEquals(Object expected, Object actual, String msg) {
|
||||
if (!expected.equals(actual))
|
||||
throw new AssertionError(msg + ": expected=" + expected + " actual=" + actual);
|
||||
}
|
||||
|
||||
static void assertTrue(boolean cond, String msg) {
|
||||
if (!cond) throw new AssertionError(msg);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
System.out.println("=== doris-0001: BindExprGroupingAlgorithm ===");
|
||||
|
||||
DefectiveAlgorithm defAlg = new DefectiveAlgorithm();
|
||||
FixedAlgorithm fixAlg = new FixedAlgorithm();
|
||||
|
||||
// Correctness: small case
|
||||
{
|
||||
int P = 10, G = 4;
|
||||
List<Expression> projections = makeProjections(P, G);
|
||||
List<Expression> grouping = makeGroupingList(G);
|
||||
|
||||
Result dr = defAlg.processNonStandardAggregate(projections, grouping);
|
||||
Result fr = fixAlg.processNonStandardAggregate(projections, grouping);
|
||||
|
||||
assertEquals(dr.output, fr.output, "output labels");
|
||||
// First G should be "keep:", rest "alias:"
|
||||
for (int i = 0; i < G; i++) assertTrue(dr.output.get(i).startsWith("keep:"), "keep at " + i);
|
||||
for (int i = G; i < P; i++) assertTrue(dr.output.get(i).startsWith("alias:"), "alias at " + i);
|
||||
System.out.println("PASS correctness (P=10, G=4)");
|
||||
}
|
||||
|
||||
// Correctness: Set passed directly (should not double-wrap)
|
||||
{
|
||||
int P = 8, G = 3;
|
||||
List<Expression> projections = makeProjections(P, G);
|
||||
Set<Expression> groupingSet = new HashSet<>(makeGroupingList(G));
|
||||
|
||||
Result dr = defAlg.processNonStandardAggregate(projections, new ArrayList<>(groupingSet));
|
||||
Result fr = fixAlg.processNonStandardAggregate(projections, groupingSet);
|
||||
assertEquals(dr.output, fr.output, "Set input output labels");
|
||||
System.out.println("PASS correctness Set input (P=8, G=3)");
|
||||
}
|
||||
|
||||
// Benchmark at P=400, G=200
|
||||
{
|
||||
int P = 400, G = 200;
|
||||
List<Expression> projections = makeProjections(P, G);
|
||||
List<Expression> grouping = makeGroupingList(G);
|
||||
|
||||
// warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defAlg.processNonStandardAggregate(projections, new ArrayList<>(grouping));
|
||||
fixAlg.processNonStandardAggregate(projections, new ArrayList<>(grouping));
|
||||
}
|
||||
|
||||
long defNs = 0, fixNs = 0;
|
||||
int reps = 50;
|
||||
for (int i = 0; i < reps; i++) {
|
||||
defNs += defAlg.processNonStandardAggregate(projections, new ArrayList<>(grouping)).ns;
|
||||
fixNs += fixAlg.processNonStandardAggregate(projections, new ArrayList<>(grouping)).ns;
|
||||
}
|
||||
defNs /= reps; fixNs /= reps;
|
||||
double ratio = (double) defNs / Math.max(1, fixNs);
|
||||
|
||||
System.out.printf("BENCH P=%d G=%d reps=%d%n", P, G, reps);
|
||||
System.out.printf(" defective avg: %,d ns%n", defNs);
|
||||
System.out.printf(" fixed avg: %,d ns%n", fixNs);
|
||||
System.out.printf(" speedup: %.1fx%n", ratio);
|
||||
|
||||
assertTrue(ratio >= 2.0, "Expected fixed >= 2x faster, got " + ratio + "x");
|
||||
System.out.println("PASS speedup >= 2x");
|
||||
}
|
||||
|
||||
System.out.println("=== ALL PASS ===");
|
||||
}
|
||||
}
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
BIN
defects/doris/unit/unit/BindExprGroupingAlgorithm$Result.class
Normal file
BIN
defects/doris/unit/unit/BindExprGroupingAlgorithm$Result.class
Normal file
Binary file not shown.
Binary file not shown.
BIN
defects/doris/unit/unit/BindExprGroupingAlgorithm.class
Normal file
BIN
defects/doris/unit/unit/BindExprGroupingAlgorithm.class
Normal file
Binary file not shown.
|
|
@ -0,0 +1,79 @@
|
|||
# foundationdb-0001 — canLaunchSrc: std::count nested inside O(S×R) double loop
|
||||
|
||||
**Project:** FoundationDB
|
||||
**File:** `fdbserver/datadistributor/DDRelocationQueue.actor.cpp`
|
||||
**Function:** `canLaunchSrc`
|
||||
**Severity:** MEDIUM
|
||||
**CWE:** CWE-407 (Inefficient Algorithmic Complexity)
|
||||
|
||||
## Defect
|
||||
|
||||
`canLaunchSrc` checks whether a data relocation can be launched without overloading
|
||||
source servers. The outer loop iterates over `relocation.src` (team size, S = 3–5
|
||||
storage servers). The inner loop iterates over `cancellableRelocations` (the queue of
|
||||
in-flight cancellable moves, R entries). Inside the inner loop, `std::count` performs
|
||||
an O(S') linear scan of each cancellable relocation's `src` vector (also of size ~S'):
|
||||
|
||||
```cpp
|
||||
for (int i = 0; i < relocation.src.size(); i++) { // O(S)
|
||||
auto busyCopy = busymap[relocation.src[i]];
|
||||
for (int j = 0; j < cancellableRelocations.size(); j++) { // O(R)
|
||||
auto& servers = cancellableRelocations[j].src;
|
||||
if (std::count(servers.begin(), servers.end(), relocation.src[i])) // O(S')
|
||||
busyCopy.removeWork(...);
|
||||
}
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
Total: **O(S × R × S')** where R grows with the number of concurrent relocations.
|
||||
In a large cluster undergoing rebalancing, R can reach hundreds or thousands,
|
||||
making this O(n²) in the relocation queue depth.
|
||||
|
||||
## Fix
|
||||
|
||||
Pre-build an `unordered_set<UID>` for each cancellable relocation's `src` servers
|
||||
before the outer loop (or use a flat `unordered_map<UID, work_to_cancel>` indexed
|
||||
by server). The inner loop then becomes an O(1) map lookup.
|
||||
|
||||
```cpp
|
||||
// Build per-server cancellable work map once
|
||||
std::unordered_map<UID, std::vector<int>> cancellable_by_server;
|
||||
for (int j = 0; j < cancellableRelocations.size(); j++) {
|
||||
for (const auto& uid : cancellableRelocations[j].src) {
|
||||
cancellable_by_server[uid].push_back(j);
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < relocation.src.size(); i++) {
|
||||
auto busyCopy = busymap[relocation.src[i]];
|
||||
// O(1) map lookup instead of O(R * S') double scan
|
||||
auto it = cancellable_by_server.find(relocation.src[i]);
|
||||
if (it != cancellable_by_server.end()) {
|
||||
for (int j : it->second) {
|
||||
busyCopy.removeWork(cancellableRelocations[j].priority,
|
||||
cancellableRelocations[j].workFactor);
|
||||
}
|
||||
}
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
Complexity: **O(S + R×S' + S)** = **O(R×S')** build + **O(S)** lookup, dominated by
|
||||
the one-time build. This removes the multiplicative factor between S and the inner
|
||||
O(R×S') scan.
|
||||
|
||||
## Impact
|
||||
|
||||
During cluster rebalancing or after storage server failures, `canLaunchSrc` is called
|
||||
for every candidate relocation in the queue (line 1090, inside a hot loop). As the
|
||||
queue grows, each call degrades from O(R) to O(S×R×S') causing compounding slowdown
|
||||
in data distribution scheduling decisions.
|
||||
|
||||
## Location
|
||||
|
||||
```
|
||||
fdbserver/datadistributor/DDRelocationQueue.actor.cpp
|
||||
canLaunchSrc() lines 436–477
|
||||
call site line 1090
|
||||
```
|
||||
239
defects/foundationdb/unit/FoundationdbTest.java
Normal file
239
defects/foundationdb/unit/FoundationdbTest.java
Normal file
|
|
@ -0,0 +1,239 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* foundationdb-0001: canLaunchSrc std::count nested loop O(S × R × S')
|
||||
*
|
||||
* Models the data structures in DDRelocationQueue.actor.cpp:
|
||||
* outer loop: relocation.src (S servers)
|
||||
* inner loop: cancellableRelocations (R entries)
|
||||
* innermost: std::count on servers vector (S')
|
||||
*
|
||||
* Compile: javac -d . FoundationdbTest.java
|
||||
* Run: java unit.FoundationdbTest
|
||||
*/
|
||||
public class FoundationdbTest {
|
||||
|
||||
// Simulate a UID (storage server ID)
|
||||
static class UID {
|
||||
final long v;
|
||||
UID(long v) { this.v = v; }
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof UID && ((UID)o).v == v;
|
||||
}
|
||||
@Override public int hashCode() { return Long.hashCode(v); }
|
||||
@Override public String toString() { return "UID(" + v + ")"; }
|
||||
}
|
||||
|
||||
// Simulate RelocateData.src (list of source server UIDs)
|
||||
static class RelocateData {
|
||||
final List<UID> src;
|
||||
final int priority;
|
||||
final int workFactor;
|
||||
RelocateData(List<UID> src, int priority, int workFactor) {
|
||||
this.src = src; this.priority = priority; this.workFactor = workFactor;
|
||||
}
|
||||
}
|
||||
|
||||
// Simulate Busyness.addWork / removeWork
|
||||
static class Busyness {
|
||||
int load = 0;
|
||||
Busyness(int load) { this.load = load; }
|
||||
Busyness copy() { return new Busyness(load); }
|
||||
void removeWork(int priority, int workFactor) { load = Math.max(0, load - workFactor); }
|
||||
boolean canLaunch(int priority, int workFactor) { return load + workFactor <= 100; }
|
||||
}
|
||||
|
||||
// ---- DEFECTIVE implementation (O(S × R × S')) ----
|
||||
static boolean canLaunchSrc_defective(
|
||||
RelocateData relocation,
|
||||
Map<UID, Busyness> busymap,
|
||||
List<RelocateData> cancellableRelocations,
|
||||
int[] comparisonCount) {
|
||||
int workFactor = 10;
|
||||
int neededServers = Math.max(1, relocation.src.size() - 3 + 1);
|
||||
|
||||
for (int i = 0; i < relocation.src.size(); i++) { // O(S)
|
||||
Busyness busyCopy = busymap.get(relocation.src.get(i)).copy();
|
||||
for (int j = 0; j < cancellableRelocations.size(); j++) { // O(R)
|
||||
List<UID> servers = cancellableRelocations.get(j).src;
|
||||
// std::count equivalent — O(S') linear scan
|
||||
boolean found = false;
|
||||
for (UID uid : servers) { // O(S')
|
||||
comparisonCount[0]++;
|
||||
if (uid.equals(relocation.src.get(i))) { found = true; break; }
|
||||
}
|
||||
if (found) {
|
||||
busyCopy.removeWork(cancellableRelocations.get(j).priority,
|
||||
cancellableRelocations.get(j).workFactor);
|
||||
}
|
||||
}
|
||||
if (busyCopy.canLaunch(relocation.priority, workFactor)) {
|
||||
--neededServers;
|
||||
if (neededServers == 0) return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// ---- FIXED implementation: pre-build map, O(R×S' + S) ----
|
||||
static boolean canLaunchSrc_fixed(
|
||||
RelocateData relocation,
|
||||
Map<UID, Busyness> busymap,
|
||||
List<RelocateData> cancellableRelocations,
|
||||
int[] comparisonCount) {
|
||||
int workFactor = 10;
|
||||
int neededServers = Math.max(1, relocation.src.size() - 3 + 1);
|
||||
|
||||
// Build inverted index: UID -> list of cancellable relocation indices O(R×S')
|
||||
Map<UID, List<Integer>> cancellableByServer = new HashMap<>();
|
||||
for (int j = 0; j < cancellableRelocations.size(); j++) {
|
||||
for (UID uid : cancellableRelocations.get(j).src) {
|
||||
comparisonCount[0]++;
|
||||
cancellableByServer.computeIfAbsent(uid, k -> new ArrayList<>()).add(j);
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < relocation.src.size(); i++) { // O(S)
|
||||
Busyness busyCopy = busymap.get(relocation.src.get(i)).copy();
|
||||
List<Integer> indices = cancellableByServer.get(relocation.src.get(i));
|
||||
if (indices != null) {
|
||||
for (int j : indices) {
|
||||
busyCopy.removeWork(cancellableRelocations.get(j).priority,
|
||||
cancellableRelocations.get(j).workFactor);
|
||||
}
|
||||
}
|
||||
if (busyCopy.canLaunch(relocation.priority, workFactor)) {
|
||||
--neededServers;
|
||||
if (neededServers == 0) return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Build a test scenario
|
||||
static Map<UID, Busyness> buildBusymap(List<UID> allServers) {
|
||||
Map<UID, Busyness> map = new HashMap<>();
|
||||
for (UID uid : allServers) map.put(uid, new Busyness(50));
|
||||
return map;
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
int pass = 0, fail = 0;
|
||||
|
||||
// -- Test 1: correctness — small scenario
|
||||
{
|
||||
List<UID> allServers = new ArrayList<>();
|
||||
for (int i = 0; i < 5; i++) allServers.add(new UID(i));
|
||||
|
||||
// The relocation being considered
|
||||
RelocateData candidate = new RelocateData(
|
||||
Arrays.asList(allServers.get(0), allServers.get(1), allServers.get(2)),
|
||||
100, 10);
|
||||
|
||||
// Some cancellable in-flight moves that involve the same servers
|
||||
List<RelocateData> cancellable = new ArrayList<>();
|
||||
cancellable.add(new RelocateData(Arrays.asList(allServers.get(0), allServers.get(3)), 50, 30));
|
||||
cancellable.add(new RelocateData(Arrays.asList(allServers.get(1), allServers.get(4)), 50, 30));
|
||||
|
||||
Map<UID, Busyness> busymap = buildBusymap(allServers);
|
||||
|
||||
int[] cmpDefective = {0};
|
||||
int[] cmpFixed = {0};
|
||||
boolean resultDef = canLaunchSrc_defective(candidate, busymap, cancellable, cmpDefective);
|
||||
boolean resultFix = canLaunchSrc_fixed(candidate, busymap, cancellable, cmpFixed);
|
||||
|
||||
if (resultDef == resultFix) {
|
||||
System.out.printf("PASS test1: correctness — defective=%b fixed=%b%n", resultDef, resultFix);
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test1: defective=%b fixed=%b%n", resultDef, resultFix);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
// -- Test 2: complexity — large R queue demonstrates O(S×R×S') vs O(R×S' + S)
|
||||
// Use servers at max load so canLaunch always fails, forcing the full loop
|
||||
{
|
||||
int S = 3; // team size (relocation.src)
|
||||
int R = 500; // queue depth (cancellable relocations)
|
||||
int S2 = 3; // src team size of each cancellable relocation
|
||||
|
||||
// Need enough servers: 3 (candidate) + R*S2 (cancellable) = 3 + 500*3 = 1503
|
||||
List<UID> servers = new ArrayList<>();
|
||||
for (int i = 0; i < 3000; i++) servers.add(new UID(i));
|
||||
|
||||
RelocateData candidate = new RelocateData(
|
||||
servers.subList(0, S), 100, 10);
|
||||
|
||||
// cancellable relocations — use disjoint servers so no work is removed
|
||||
// and servers stay busy, forcing all S iterations to complete
|
||||
List<RelocateData> cancellable = new ArrayList<>();
|
||||
for (int j = 0; j < R; j++) {
|
||||
int base = 100 + j * S2; // disjoint from candidate servers (0,1,2)
|
||||
cancellable.add(new RelocateData(
|
||||
servers.subList(base, base + S2), 50, 5));
|
||||
}
|
||||
|
||||
// Overload the source servers so canLaunch never returns true early
|
||||
Map<UID, Busyness> busymapDef = new HashMap<>();
|
||||
Map<UID, Busyness> busymapFix = new HashMap<>();
|
||||
for (UID uid : servers) {
|
||||
busymapDef.put(uid, new Busyness(95)); // high load — can't launch
|
||||
busymapFix.put(uid, new Busyness(95));
|
||||
}
|
||||
|
||||
int[] cmpDefective = {0};
|
||||
int[] cmpFixed = {0};
|
||||
|
||||
boolean resDef = canLaunchSrc_defective(candidate, busymapDef, cancellable, cmpDefective);
|
||||
boolean resFix = canLaunchSrc_fixed(candidate, busymapFix, cancellable, cmpFixed);
|
||||
|
||||
System.out.printf("test2: defective comparisons=%d fixed comparisons=%d (S=%d R=%d S2=%d)%n",
|
||||
cmpDefective[0], cmpFixed[0], S, R, S2);
|
||||
System.out.printf("test2: defective result=%b fixed result=%b%n", resDef, resFix);
|
||||
|
||||
// Fixed must produce same result with fewer comparisons (no S multiplier)
|
||||
// Defective: S * R * S2 = 3 * 500 * 3 = 4500 comparisons
|
||||
// Fixed: R * S2 (build) = 500 * 3 = 1500 comparisons (no outer S factor)
|
||||
boolean sameResult = (resDef == resFix);
|
||||
boolean moreEfficient = cmpFixed[0] < cmpDefective[0];
|
||||
if (sameResult && moreEfficient) {
|
||||
System.out.println("PASS test2: fixed is more efficient and produces same result");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test2: sameResult=%b moreEfficient=%b%n", sameResult, moreEfficient);
|
||||
fail++;
|
||||
}
|
||||
|
||||
int expectedDefective = S * R * S2;
|
||||
int expectedFixed = R * S2; // build phase dominates
|
||||
System.out.printf(" Expected defective ~O(S×R×S')=%d, actual=%d%n", expectedDefective, cmpDefective[0]);
|
||||
System.out.printf(" Expected fixed ~O(R×S')=%d, actual=%d%n", expectedFixed, cmpFixed[0]);
|
||||
}
|
||||
|
||||
// -- Test 3: no cancellable relocations — both return same result
|
||||
{
|
||||
List<UID> servers = new ArrayList<>();
|
||||
for (int i = 0; i < 3; i++) servers.add(new UID(i));
|
||||
RelocateData candidate = new RelocateData(servers, 100, 5);
|
||||
Map<UID, Busyness> busymap = buildBusymap(servers);
|
||||
|
||||
int[] c1 = {0}, c2 = {0};
|
||||
boolean r1 = canLaunchSrc_defective(candidate, busymap, new ArrayList<>(), c1);
|
||||
boolean r2 = canLaunchSrc_fixed(candidate, busymap, new ArrayList<>(), c2);
|
||||
|
||||
if (r1 == r2) {
|
||||
System.out.printf("PASS test3: empty cancellable — both=%b%n", r1);
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test3: mismatch on empty cancellable%n");
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf("%nResults: %d passed, %d failed%n", pass, fail);
|
||||
if (fail > 0) System.exit(1);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
# kylin-0001: JdbcJobScheduler — List<String>.contains in stream filter → O(J²)
|
||||
|
||||
## Classification
|
||||
- **Severity**: MEDIUM
|
||||
- **CWE**: CWE-407 (Algorithmic Complexity — Inefficient Algorithmic Complexity)
|
||||
- **Component**: `src/core-job/src/main/java/org/apache/kylin/job/scheduler/JdbcJobScheduler.java`
|
||||
- **Method**: `releaseJobLock()` (internal scheduler loop)
|
||||
|
||||
## Defect
|
||||
|
||||
In `releaseJobLock()` (approximate method name from context around line 410–432):
|
||||
|
||||
```java
|
||||
List<String> jobInfoIds = jobs.stream()
|
||||
.map(JobInfo::getJobId)
|
||||
.collect(Collectors.toList()); // <-- ArrayList
|
||||
|
||||
List<String> toRemoveLocks = Lists.newArrayList(jobIds).stream()
|
||||
.filter(jobId -> !jobInfoIds.contains(jobId)) // <-- O(J) List.contains
|
||||
.collect(Collectors.toList());
|
||||
```
|
||||
|
||||
`jobInfoIds` is collected into an `ArrayList<String>`. The stream filter calls
|
||||
`jobInfoIds.contains(jobId)` once per element of `jobIds`. Both lists grow with the number of
|
||||
in-flight jobs. Total: **O(J²)** where J = number of job IDs.
|
||||
|
||||
This scheduler loop runs on a recurring timer. Under high-throughput build workloads (large Kylin
|
||||
clusters doing continuous incremental cube builds), J can grow into the hundreds per batch, causing
|
||||
quadratic behavior in the scheduler hot path.
|
||||
|
||||
## Root Cause
|
||||
|
||||
`Collectors.toList()` produces an `ArrayList`. The calling code immediately uses it for membership
|
||||
tests only. No `Set` conversion was applied.
|
||||
|
||||
## Fix
|
||||
|
||||
Collect into a `HashSet` (or convert before use):
|
||||
|
||||
```java
|
||||
// Before:
|
||||
List<String> jobInfoIds = jobs.stream()
|
||||
.map(JobInfo::getJobId)
|
||||
.collect(Collectors.toList());
|
||||
|
||||
// After:
|
||||
Set<String> jobInfoIds = jobs.stream()
|
||||
.map(JobInfo::getJobId)
|
||||
.collect(Collectors.toCollection(HashSet::new));
|
||||
```
|
||||
|
||||
## Complexity
|
||||
|
||||
| Before | After |
|
||||
|--------|-------|
|
||||
| O(J²) per scheduler tick | O(J) per scheduler tick |
|
||||
|
||||
With J=200 jobs per batch: Before = 40 000 comparisons. After = 200. **200× reduction**.
|
||||
|
||||
## Patch
|
||||
|
||||
```diff
|
||||
--- a/src/core-job/src/main/java/org/apache/kylin/job/scheduler/JdbcJobScheduler.java
|
||||
+++ b/src/core-job/src/main/java/org/apache/kylin/job/scheduler/JdbcJobScheduler.java
|
||||
@@ -415,7 +415,8 @@ public class JdbcJobScheduler {
|
||||
filter.setJobIds(jobIds);
|
||||
List<JobInfo> jobs = jobContext.getJobInfoMapper().selectByJobFilter(filter);
|
||||
- List<String> jobInfoIds = jobs.stream().map(JobInfo::getJobId).collect(Collectors.toList());
|
||||
+ // Use HashSet for O(1) membership test in stream filter below
|
||||
+ Set<String> jobInfoIds = jobs.stream().map(JobInfo::getJobId)
|
||||
+ .collect(Collectors.toCollection(HashSet::new));
|
||||
List<String> toRemoveLocks = Lists.newArrayList(jobIds).stream()
|
||||
.filter(jobId -> !jobInfoIds.contains(jobId))
|
||||
.collect(Collectors.toList());
|
||||
```
|
||||
148
defects/kylin/unit/JobSchedulerAlgorithm.java
Normal file
148
defects/kylin/unit/JobSchedulerAlgorithm.java
Normal file
|
|
@ -0,0 +1,148 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.*;
|
||||
|
||||
/**
|
||||
* Standalone unit test for kylin-0001:
|
||||
* JdbcJobScheduler — List<String>.contains() in stream filter → O(J²).
|
||||
*
|
||||
* Simulates the releaseJobLock() pattern:
|
||||
* jobInfoIds collected as List<String>, then used in stream filter with .contains().
|
||||
*
|
||||
* Compile: javac -d . JobSchedulerAlgorithm.java
|
||||
* Run: java unit.JobSchedulerAlgorithm
|
||||
*/
|
||||
public class JobSchedulerAlgorithm {
|
||||
|
||||
// ── Result ───────────────────────────────────────────────────────────────
|
||||
|
||||
static class Result {
|
||||
final List<String> toRemoveLocks;
|
||||
final long ns;
|
||||
Result(List<String> locks, long ns) { this.toRemoveLocks = locks; this.ns = ns; }
|
||||
}
|
||||
|
||||
// ── Defective: collects jobInfoIds into List → O(J) .contains per jobId ─
|
||||
|
||||
static class DefectiveAlgorithm {
|
||||
/**
|
||||
* @param jobIds all job IDs found in the lock table
|
||||
* @param jobInfoIds job IDs that have corresponding JobInfo records (as List)
|
||||
*/
|
||||
Result releaseJobLock(List<String> jobIds, List<String> jobInfoIds) {
|
||||
long t0 = System.nanoTime();
|
||||
// Mirrors the defective code:
|
||||
// List<String> jobInfoIds = jobs.stream().map(JobInfo::getJobId).collect(Collectors.toList());
|
||||
// ... filter(jobId -> !jobInfoIds.contains(jobId)) ...
|
||||
List<String> toRemoveLocks = jobIds.stream()
|
||||
.filter(jobId -> !jobInfoIds.contains(jobId)) // CWE-407 site: O(J) per call
|
||||
.collect(Collectors.toList());
|
||||
return new Result(toRemoveLocks, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Fixed: uses Set<String> → O(1) .contains ────────────────────────────
|
||||
|
||||
static class FixedAlgorithm {
|
||||
Result releaseJobLock(List<String> jobIds, List<String> jobInfoIdsList) {
|
||||
long t0 = System.nanoTime();
|
||||
// Fix: collect into HashSet for O(1) membership
|
||||
Set<String> jobInfoIds = new HashSet<>(jobInfoIdsList);
|
||||
List<String> toRemoveLocks = jobIds.stream()
|
||||
.filter(jobId -> !jobInfoIds.contains(jobId)) // O(1)
|
||||
.collect(Collectors.toList());
|
||||
return new Result(toRemoveLocks, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
static List<String> makeJobIds(int N) {
|
||||
List<String> ids = new ArrayList<>(N);
|
||||
for (int i = 0; i < N; i++) ids.add("job-" + i);
|
||||
return ids;
|
||||
}
|
||||
|
||||
// ── Assertions ───────────────────────────────────────────────────────────
|
||||
|
||||
static void assertEquals(Object expected, Object actual, String msg) {
|
||||
if (!expected.equals(actual))
|
||||
throw new AssertionError(msg + ": expected=" + expected + " actual=" + actual);
|
||||
}
|
||||
|
||||
static void assertTrue(boolean cond, String msg) {
|
||||
if (!cond) throw new AssertionError(msg);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
System.out.println("=== kylin-0001: JobSchedulerAlgorithm ===");
|
||||
|
||||
DefectiveAlgorithm defAlg = new DefectiveAlgorithm();
|
||||
FixedAlgorithm fixAlg = new FixedAlgorithm();
|
||||
|
||||
// Correctness test 1: some jobs have info, some don't
|
||||
{
|
||||
List<String> jobIds = makeJobIds(20);
|
||||
// Only even-numbered jobs have a JobInfo record
|
||||
List<String> jobInfoIds = jobIds.stream()
|
||||
.filter(id -> Integer.parseInt(id.replace("job-", "")) % 2 == 0)
|
||||
.collect(Collectors.toList());
|
||||
|
||||
Result dr = defAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds));
|
||||
Result fr = fixAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds));
|
||||
|
||||
// Sort both for comparison (order may differ)
|
||||
List<String> dSorted = new ArrayList<>(dr.toRemoveLocks); Collections.sort(dSorted);
|
||||
List<String> fSorted = new ArrayList<>(fr.toRemoveLocks); Collections.sort(fSorted);
|
||||
|
||||
assertEquals(dSorted, fSorted, "toRemoveLocks contents");
|
||||
// All odd-numbered jobs should be in toRemoveLocks
|
||||
assertTrue(dr.toRemoveLocks.size() == 10, "expected 10 orphan locks, got " + dr.toRemoveLocks.size());
|
||||
System.out.println("PASS correctness (J=20, half missing)");
|
||||
}
|
||||
|
||||
// Correctness test 2: all jobs have info records (no locks to remove)
|
||||
{
|
||||
List<String> jobIds = makeJobIds(10);
|
||||
Result dr = defAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobIds));
|
||||
Result fr = fixAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobIds));
|
||||
assertEquals(0, dr.toRemoveLocks.size(), "no orphan locks (defective)");
|
||||
assertEquals(0, fr.toRemoveLocks.size(), "no orphan locks (fixed)");
|
||||
System.out.println("PASS correctness all-present (J=10)");
|
||||
}
|
||||
|
||||
// Benchmark at J=2000
|
||||
{
|
||||
int J = 2000;
|
||||
List<String> jobIds = makeJobIds(J);
|
||||
// Half the jobs have info records
|
||||
List<String> jobInfoIds = jobIds.subList(0, J / 2);
|
||||
|
||||
// warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds));
|
||||
fixAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds));
|
||||
}
|
||||
|
||||
long defNs = 0, fixNs = 0;
|
||||
int reps = 30;
|
||||
for (int i = 0; i < reps; i++) {
|
||||
defNs += defAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds)).ns;
|
||||
fixNs += fixAlg.releaseJobLock(new ArrayList<>(jobIds), new ArrayList<>(jobInfoIds)).ns;
|
||||
}
|
||||
defNs /= reps; fixNs /= reps;
|
||||
double ratio = (double) defNs / Math.max(1, fixNs);
|
||||
|
||||
System.out.printf("BENCH J=%d reps=%d%n", J, reps);
|
||||
System.out.printf(" defective avg: %,d ns%n", defNs);
|
||||
System.out.printf(" fixed avg: %,d ns%n", fixNs);
|
||||
System.out.printf(" speedup: %.1fx%n", ratio);
|
||||
|
||||
assertTrue(ratio >= 2.0, "Expected fixed >= 2x faster, got " + ratio + "x");
|
||||
System.out.println("PASS speedup >= 2x");
|
||||
}
|
||||
|
||||
System.out.println("=== ALL PASS ===");
|
||||
}
|
||||
}
|
||||
Binary file not shown.
Binary file not shown.
BIN
defects/kylin/unit/unit/JobSchedulerAlgorithm$Result.class
Normal file
BIN
defects/kylin/unit/unit/JobSchedulerAlgorithm$Result.class
Normal file
Binary file not shown.
BIN
defects/kylin/unit/unit/JobSchedulerAlgorithm.class
Normal file
BIN
defects/kylin/unit/unit/JobSchedulerAlgorithm.class
Normal file
Binary file not shown.
|
|
@ -0,0 +1,111 @@
|
|||
# scylladb-0001 — storage_proxy::intersection: O(L₁×L₂) linear scan in vnode range loop
|
||||
|
||||
**Project:** ScyllaDB
|
||||
**File:** `service/storage_proxy.cc`
|
||||
**Function:** `storage_proxy::intersection` / range-merge loop in `query_ranges_to_vnodes`
|
||||
**Severity:** MEDIUM
|
||||
**CWE:** CWE-407 (Inefficient Algorithmic Complexity)
|
||||
|
||||
## Defect
|
||||
|
||||
`storage_proxy::intersection` computes the set intersection of two
|
||||
`host_id_vector_replica_set` vectors using `std::remove_copy_if` with a lambda
|
||||
that calls `std::find` on `l2` for each element of `l1`:
|
||||
|
||||
```cpp
|
||||
host_id_vector_replica_set storage_proxy::intersection(
|
||||
const host_id_vector_replica_set& l1,
|
||||
const host_id_vector_replica_set& l2) {
|
||||
host_id_vector_replica_set inter;
|
||||
inter.reserve(l1.size());
|
||||
std::remove_copy_if(l1.begin(), l1.end(), std::back_inserter(inter),
|
||||
[&l2] (const locator::host_id& a) {
|
||||
return std::find(l2.begin(), l2.end(), a) == l2.end(); // O(|l2|)
|
||||
});
|
||||
return inter;
|
||||
}
|
||||
```
|
||||
|
||||
This function is called **twice per vnode** inside the range-merge loop at
|
||||
`storage_proxy.cc` lines 6473–6474:
|
||||
|
||||
```cpp
|
||||
while (i != ranges.end()) { // O(V) — iterates vnodes being merged
|
||||
...
|
||||
host_id_vector_replica_set merged =
|
||||
intersection(live_endpoints, next_endpoints); // O(RF²)
|
||||
host_id_vector_replica_set current_merged_preferred =
|
||||
intersection(merged_preferred_replicas,
|
||||
current_range_preferred_replicas); // O(RF²)
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
Total per range scan: **O(V × RF²)** where:
|
||||
- V = number of vnodes being merged (up to `num_tokens` per node, typically 256)
|
||||
- RF = replication factor (typically 3)
|
||||
|
||||
With RF=3 the constant is small, but with larger RF values (RF=5 or RF=7 in
|
||||
high-availability configs) and high vnode counts (256), the quadratic factor in RF
|
||||
becomes measurable: 256 × 49 = 12,544 comparisons per scan where 256×3 = 768 suffice.
|
||||
For EACH partition key scatter/gather query, this loop runs for the token range
|
||||
covered by the query.
|
||||
|
||||
## Fix
|
||||
|
||||
Replace the linear `std::find` with an `unordered_set` for O(1) lookup. Because RF
|
||||
is typically small (3–7), build the set from the smaller operand:
|
||||
|
||||
```cpp
|
||||
host_id_vector_replica_set storage_proxy::intersection(
|
||||
const host_id_vector_replica_set& l1,
|
||||
const host_id_vector_replica_set& l2) {
|
||||
// Build hash set from the smaller set — O(min(|l1|,|l2|))
|
||||
std::unordered_set<locator::host_id> s2(l2.begin(), l2.end());
|
||||
host_id_vector_replica_set inter;
|
||||
inter.reserve(std::min(l1.size(), l2.size()));
|
||||
for (const auto& a : l1) {
|
||||
if (s2.count(a)) {
|
||||
inter.push_back(a);
|
||||
}
|
||||
}
|
||||
return inter;
|
||||
}
|
||||
```
|
||||
|
||||
Complexity: **O(|l1| + |l2|)** instead of **O(|l1| × |l2|)**.
|
||||
|
||||
Alternatively, since both vectors are small and bounded by RF, use `std::sort` +
|
||||
`std::set_intersection` for a cache-friendly O(RF log RF) solution that avoids
|
||||
hash overhead:
|
||||
|
||||
```cpp
|
||||
// Sort copies and use linear set_intersection
|
||||
auto sorted1 = l1; std::sort(sorted1.begin(), sorted1.end());
|
||||
auto sorted2 = l2; std::sort(sorted2.begin(), sorted2.end());
|
||||
host_id_vector_replica_set inter;
|
||||
std::set_intersection(sorted1.begin(), sorted1.end(),
|
||||
sorted2.begin(), sorted2.end(),
|
||||
std::back_inserter(inter));
|
||||
return inter;
|
||||
```
|
||||
|
||||
## Impact
|
||||
|
||||
Every range query that crosses multiple vnodes (which includes all multi-key batch
|
||||
reads and range scans on vnode clusters) invokes this function twice per vnode.
|
||||
Under load with 256 vnodes and RF=5, each range scan does 512 quadratic intersection
|
||||
calls. This contributes to latency spikes during cluster rebalancing when the
|
||||
range-merge logic is most active.
|
||||
|
||||
Note: clusters using tablets (the newer ScyllaDB topology) bypass this path via the
|
||||
`!erm->get_replication_strategy().uses_tablets()` guard at line 6432, so this defect
|
||||
is specific to the legacy vnode replication path.
|
||||
|
||||
## Location
|
||||
|
||||
```
|
||||
service/storage_proxy.cc
|
||||
intersection() lines 7135–7142
|
||||
vnode range-merge loop lines 6418–6474 (calls intersection at 6473–6474)
|
||||
```
|
||||
215
defects/scylladb/unit/ScylladbTest.java
Normal file
215
defects/scylladb/unit/ScylladbTest.java
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* scylladb-0001: storage_proxy::intersection O(|l1|×|l2|) linear scan in vnode range loop
|
||||
*
|
||||
* Models service/storage_proxy.cc:
|
||||
* intersection(): std::remove_copy_if with std::find on l2 — O(|l1| × |l2|)
|
||||
* vnode range-merge loop: calls intersection twice per vnode — O(V × RF²)
|
||||
*
|
||||
* Compile: javac -d . ScylladbTest.java
|
||||
* Run: java unit.ScylladbTest
|
||||
*/
|
||||
public class ScylladbTest {
|
||||
|
||||
static class HostId {
|
||||
final long id;
|
||||
HostId(long id) { this.id = id; }
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof HostId && ((HostId)o).id == id;
|
||||
}
|
||||
@Override public int hashCode() { return Long.hashCode(id); }
|
||||
@Override public String toString() { return "H" + id; }
|
||||
}
|
||||
|
||||
// ---- DEFECTIVE: O(|l1| × |l2|) ----
|
||||
static List<HostId> intersection_defective(
|
||||
List<HostId> l1, List<HostId> l2, int[] comparisonCount) {
|
||||
List<HostId> result = new ArrayList<>();
|
||||
for (HostId a : l1) {
|
||||
// std::find on l2 — O(|l2|)
|
||||
boolean found = false;
|
||||
for (HostId b : l2) {
|
||||
comparisonCount[0]++;
|
||||
if (a.equals(b)) { found = true; break; }
|
||||
}
|
||||
if (found) result.add(a);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// Simulate vnode range-merge loop — calls intersection twice per vnode
|
||||
static int vnodeRangeMerge_defective(
|
||||
List<List<HostId>> vnodeLiveEndpoints, // one replica set per vnode
|
||||
List<List<HostId>> vnodePreferredEndpoints,
|
||||
int[] comparisonCount) {
|
||||
int mergedRanges = 0;
|
||||
List<HostId> mergedLive = vnodeLiveEndpoints.get(0);
|
||||
List<HostId> mergedPreferred = vnodePreferredEndpoints.get(0);
|
||||
|
||||
for (int i = 1; i < vnodeLiveEndpoints.size(); i++) {
|
||||
List<HostId> nextLive = vnodeLiveEndpoints.get(i);
|
||||
List<HostId> nextPreferred = vnodePreferredEndpoints.get(i);
|
||||
|
||||
// Two intersection calls per vnode
|
||||
List<HostId> merged = intersection_defective(mergedLive, nextLive, comparisonCount);
|
||||
List<HostId> mergedPref = intersection_defective(mergedPreferred, nextPreferred, comparisonCount);
|
||||
|
||||
if (merged.size() >= 1) { // enough endpoints to satisfy CL
|
||||
mergedLive = merged;
|
||||
mergedPreferred = mergedPref;
|
||||
mergedRanges++;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return mergedRanges;
|
||||
}
|
||||
|
||||
// ---- FIXED: O(|l1| + |l2|) using HashSet ----
|
||||
static List<HostId> intersection_fixed(
|
||||
List<HostId> l1, List<HostId> l2, int[] comparisonCount) {
|
||||
Set<HostId> s2 = new HashSet<>(l2);
|
||||
comparisonCount[0] += l2.size(); // cost of building the set
|
||||
List<HostId> result = new ArrayList<>();
|
||||
for (HostId a : l1) {
|
||||
comparisonCount[0]++;
|
||||
if (s2.contains(a)) result.add(a);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
static int vnodeRangeMerge_fixed(
|
||||
List<List<HostId>> vnodeLiveEndpoints,
|
||||
List<List<HostId>> vnodePreferredEndpoints,
|
||||
int[] comparisonCount) {
|
||||
int mergedRanges = 0;
|
||||
List<HostId> mergedLive = vnodeLiveEndpoints.get(0);
|
||||
List<HostId> mergedPreferred = vnodePreferredEndpoints.get(0);
|
||||
|
||||
for (int i = 1; i < vnodeLiveEndpoints.size(); i++) {
|
||||
List<HostId> nextLive = vnodeLiveEndpoints.get(i);
|
||||
List<HostId> nextPreferred = vnodePreferredEndpoints.get(i);
|
||||
|
||||
List<HostId> merged = intersection_fixed(mergedLive, nextLive, comparisonCount);
|
||||
List<HostId> mergedPref = intersection_fixed(mergedPreferred, nextPreferred, comparisonCount);
|
||||
|
||||
if (merged.size() >= 1) {
|
||||
mergedLive = merged;
|
||||
mergedPreferred = mergedPref;
|
||||
mergedRanges++;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return mergedRanges;
|
||||
}
|
||||
|
||||
// Build vnode sets: V vnodes, RF replicas each, with rolling overlap
|
||||
static List<List<HostId>> buildVnodeEndpoints(int V, int RF, int totalNodes) {
|
||||
List<List<HostId>> result = new ArrayList<>();
|
||||
List<HostId> nodes = new ArrayList<>();
|
||||
for (int i = 0; i < totalNodes; i++) nodes.add(new HostId(i));
|
||||
|
||||
for (int v = 0; v < V; v++) {
|
||||
List<HostId> replicas = new ArrayList<>();
|
||||
for (int r = 0; r < RF; r++) {
|
||||
replicas.add(nodes.get((v + r) % totalNodes));
|
||||
}
|
||||
result.add(replicas);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
int pass = 0, fail = 0;
|
||||
|
||||
// -- Test 1: correctness — simple intersection
|
||||
{
|
||||
List<HostId> l1 = Arrays.asList(new HostId(1), new HostId(2), new HostId(3));
|
||||
List<HostId> l2 = Arrays.asList(new HostId(2), new HostId(3), new HostId(4));
|
||||
int[] c1 = {0}, c2 = {0};
|
||||
List<HostId> r1 = intersection_defective(l1, l2, c1);
|
||||
List<HostId> r2 = intersection_fixed(l1, l2, c2);
|
||||
|
||||
if (r1.equals(r2)) {
|
||||
System.out.printf("PASS test1: intersection correctness — result=%s%n", r1);
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test1: defective=%s fixed=%s%n", r1, r2);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
// -- Test 2: complexity with vnode range-merge loop
|
||||
{
|
||||
int V = 256; // vnodes (typical ScyllaDB vnode count)
|
||||
int RF = 5; // replication factor
|
||||
int N = 10; // nodes in cluster
|
||||
|
||||
List<List<HostId>> liveEps = buildVnodeEndpoints(V, RF, N);
|
||||
List<List<HostId>> prefEps = buildVnodeEndpoints(V, RF, N);
|
||||
|
||||
int[] cmpDef = {0}, cmpFix = {0};
|
||||
int mergedDef = vnodeRangeMerge_defective(liveEps, prefEps, cmpDef);
|
||||
int mergedFix = vnodeRangeMerge_fixed(liveEps, prefEps, cmpFix);
|
||||
|
||||
System.out.printf("test2: defective comparisons=%d fixed comparisons=%d (V=%d RF=%d N=%d)%n",
|
||||
cmpDef[0], cmpFix[0], V, RF, N);
|
||||
System.out.printf("test2: merged ranges — defective=%d fixed=%d%n", mergedDef, mergedFix);
|
||||
|
||||
if (mergedDef == mergedFix && cmpFix[0] < cmpDef[0]) {
|
||||
System.out.println("PASS test2: fixed is more efficient and produces same result");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test2: match=%b efficient=%b%n",
|
||||
mergedDef == mergedFix, cmpFix[0] < cmpDef[0]);
|
||||
fail++;
|
||||
}
|
||||
|
||||
// Show expected vs actual complexity
|
||||
// Defective: per vnode, 2 intersections each O(RF²) = 2 * V * RF²
|
||||
// Fixed: per vnode, 2 intersections each O(RF) = 2 * V * RF
|
||||
int expDef = 2 * V * RF * RF;
|
||||
int expFix = 2 * V * RF * 2; // build + scan
|
||||
System.out.printf(" Expected defective ~O(2×V×RF²)=%d, actual=%d%n", expDef, cmpDef[0]);
|
||||
System.out.printf(" Expected fixed ~O(2×V×RF)=%d, actual=%d%n", expFix, cmpFix[0]);
|
||||
}
|
||||
|
||||
// -- Test 3: empty intersection
|
||||
{
|
||||
List<HostId> l1 = Arrays.asList(new HostId(1), new HostId(2));
|
||||
List<HostId> l2 = Arrays.asList(new HostId(3), new HostId(4));
|
||||
int[] c1 = {0}, c2 = {0};
|
||||
List<HostId> r1 = intersection_defective(l1, l2, c1);
|
||||
List<HostId> r2 = intersection_fixed(l1, l2, c2);
|
||||
if (r1.isEmpty() && r2.isEmpty()) {
|
||||
System.out.println("PASS test3: empty intersection");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test3: def=%s fix=%s%n", r1, r2);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
// -- Test 4: full intersection (all elements common)
|
||||
{
|
||||
List<HostId> hosts = Arrays.asList(new HostId(1), new HostId(2), new HostId(3));
|
||||
int[] c1 = {0}, c2 = {0};
|
||||
List<HostId> r1 = intersection_defective(hosts, hosts, c1);
|
||||
List<HostId> r2 = intersection_fixed(hosts, hosts, c2);
|
||||
if (r1.equals(r2) && r1.equals(hosts)) {
|
||||
System.out.println("PASS test4: full intersection");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test4: def=%s fix=%s%n", r1, r2);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf("%nResults: %d passed, %d failed%n", pass, fail);
|
||||
if (fail > 0) System.exit(1);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
# starrocks-0001: MaterializedViewRewriter.getTableToRelationid — List.contains in column-ref loop → O(N×T)
|
||||
|
||||
## Classification
|
||||
- **Severity**: HIGH
|
||||
- **CWE**: CWE-407 (Algorithmic Complexity — Inefficient Algorithmic Complexity)
|
||||
- **Component**: `fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MaterializedViewRewriter.java`
|
||||
- **Method**: `getTableToRelationid()`
|
||||
|
||||
## Defect
|
||||
|
||||
`getTableToRelationid()` iterates over every column-ref-to-table mapping in the `ColumnRefFactory`
|
||||
(up to N entries for wide schemas) and calls `tableList.contains(entry.getValue())` where `tableList`
|
||||
is the `List<Table>` parameter.
|
||||
|
||||
```java
|
||||
private Map<Table, Set<Integer>> getTableToRelationid(
|
||||
OptExpression optExpression, ColumnRefFactory refFactory, List<Table> tableList) {
|
||||
...
|
||||
for (Map.Entry<ColumnRefOperator, Table> entry : refFactory.getColumnRefToTable().entrySet()) {
|
||||
if (!tableList.contains(entry.getValue())) { // <-- O(T) List.contains
|
||||
continue;
|
||||
}
|
||||
...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`List.contains()` is O(T) where T is the number of tables. The loop runs N times (one per column ref).
|
||||
Total: **O(N × T)**.
|
||||
|
||||
The method is called **twice** per MV rewrite candidate — once for the query expression and once for
|
||||
the MV expression (lines 2645 and 2656). MV rewrite is attempted for every applicable MV on every
|
||||
query, so in workloads with many columns and multiple MVs this compounds quickly.
|
||||
|
||||
A production StarRocks deployment with 500 column refs and 20 candidate tables performs 10 000
|
||||
list-scans per call, repeated for every MV candidate at query-compilation time.
|
||||
|
||||
## Root Cause
|
||||
|
||||
The caller (`rewrite()`/`generateRelationIdMap()`) has `List<Table> queryTables` / `mvTables` which
|
||||
are passed directly to `getTableToRelationid`. The lists are never converted to sets before the call.
|
||||
|
||||
## Fix
|
||||
|
||||
Convert `tableList` to a `HashSet` once before the loop. Each `.contains()` becomes O(1).
|
||||
|
||||
```java
|
||||
// Before:
|
||||
if (!tableList.contains(entry.getValue())) { // O(T) per iteration
|
||||
|
||||
// After:
|
||||
Set<Table> tableSet = new HashSet<>(tableList); // O(T) once, before loop
|
||||
...
|
||||
if (!tableSet.contains(entry.getValue())) { // O(1) per iteration
|
||||
```
|
||||
|
||||
Alternatively, change the parameter type from `List<Table>` to `Set<Table>` and update callers to
|
||||
pass `new HashSet<>(queryTables)` / `new HashSet<>(mvTables)`.
|
||||
|
||||
## Complexity
|
||||
|
||||
| Before | After |
|
||||
|--------|-------|
|
||||
| O(N × T) per call | O(N + T) per call |
|
||||
|
||||
With N=500 column refs, T=20 tables: Before = 10 000 comparisons. After = 520. **~19× reduction**.
|
||||
Called twice per MV candidate; with 10 MVs = 200 000 → 10 400 comparisons total.
|
||||
|
||||
## Patch
|
||||
|
||||
```diff
|
||||
--- a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MaterializedViewRewriter.java
|
||||
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MaterializedViewRewriter.java
|
||||
@@ -2777,9 +2777,11 @@ public class MaterializedViewRewriter {
|
||||
private Map<Table, Set<Integer>> getTableToRelationid(
|
||||
OptExpression optExpression, ColumnRefFactory refFactory, List<Table> tableList) {
|
||||
Map<Table, Set<Integer>> tableToRelationId = Maps.newHashMap();
|
||||
Set<ColumnRefOperator> validColumnRefs = MvUtils.collectScanColumn(optExpression);
|
||||
+ // Convert to HashSet once — avoids O(T) List.contains per column-ref iteration
|
||||
+ Set<Table> tableSet = new HashSet<>(tableList);
|
||||
for (Map.Entry<ColumnRefOperator, Table> entry : refFactory.getColumnRefToTable().entrySet()) {
|
||||
- if (!tableList.contains(entry.getValue())) {
|
||||
+ if (!tableSet.contains(entry.getValue())) {
|
||||
continue;
|
||||
}
|
||||
```
|
||||
177
defects/starrocks/unit/MvTableRelationAlgorithm.java
Normal file
177
defects/starrocks/unit/MvTableRelationAlgorithm.java
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.*;
|
||||
|
||||
/**
|
||||
* Standalone unit test for starrocks-0001:
|
||||
* MaterializedViewRewriter.getTableToRelationid — List.contains() in column-ref loop → O(N×T).
|
||||
*
|
||||
* Simulates getTableToRelationid():
|
||||
* for each column-ref → table entry, test membership in tableList.
|
||||
*
|
||||
* Compile: javac -d . MvTableRelationAlgorithm.java
|
||||
* Run: java unit.MvTableRelationAlgorithm
|
||||
*/
|
||||
public class MvTableRelationAlgorithm {
|
||||
|
||||
// ── Minimal stubs ────────────────────────────────────────────────────────
|
||||
|
||||
static class Table {
|
||||
final String name;
|
||||
Table(String name) { this.name = name; }
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof Table && name.equals(((Table) o).name);
|
||||
}
|
||||
@Override public int hashCode() { return name.hashCode(); }
|
||||
@Override public String toString() { return name; }
|
||||
}
|
||||
|
||||
static class ColRef {
|
||||
final int id;
|
||||
ColRef(int id) { this.id = id; }
|
||||
@Override public int hashCode() { return id; }
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof ColRef && id == ((ColRef) o).id;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Result ───────────────────────────────────────────────────────────────
|
||||
|
||||
static class Result {
|
||||
final Map<Table, Set<Integer>> tableToRelationId;
|
||||
final long ns;
|
||||
Result(Map<Table, Set<Integer>> m, long ns) { this.tableToRelationId = m; this.ns = ns; }
|
||||
}
|
||||
|
||||
// ── Defective: List<Table>.contains() O(T) per column ref ────────────────
|
||||
|
||||
static class DefectiveAlgorithm {
|
||||
Result getTableToRelationid(
|
||||
Map<ColRef, Table> colRefToTable,
|
||||
Set<ColRef> validColumnRefs,
|
||||
List<Table> tableList) { // List → O(T) .contains
|
||||
Map<Table, Set<Integer>> result = new HashMap<>();
|
||||
long t0 = System.nanoTime();
|
||||
for (Map.Entry<ColRef, Table> entry : colRefToTable.entrySet()) {
|
||||
if (!tableList.contains(entry.getValue())) { // CWE-407 site
|
||||
continue;
|
||||
}
|
||||
if (!validColumnRefs.contains(entry.getKey())) {
|
||||
continue;
|
||||
}
|
||||
result.computeIfAbsent(entry.getValue(), k -> new HashSet<>())
|
||||
.add(entry.getKey().id % 100); // synthetic relation ID
|
||||
}
|
||||
return new Result(result, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Fixed: Set<Table>.contains() O(1) ────────────────────────────────────
|
||||
|
||||
static class FixedAlgorithm {
|
||||
Result getTableToRelationid(
|
||||
Map<ColRef, Table> colRefToTable,
|
||||
Set<ColRef> validColumnRefs,
|
||||
List<Table> tableList) {
|
||||
Map<Table, Set<Integer>> result = new HashMap<>();
|
||||
long t0 = System.nanoTime();
|
||||
Set<Table> tableSet = new HashSet<>(tableList); // O(T) once
|
||||
for (Map.Entry<ColRef, Table> entry : colRefToTable.entrySet()) {
|
||||
if (!tableSet.contains(entry.getValue())) { // O(1)
|
||||
continue;
|
||||
}
|
||||
if (!validColumnRefs.contains(entry.getKey())) {
|
||||
continue;
|
||||
}
|
||||
result.computeIfAbsent(entry.getValue(), k -> new HashSet<>())
|
||||
.add(entry.getKey().id % 100);
|
||||
}
|
||||
return new Result(result, System.nanoTime() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Test data helpers ────────────────────────────────────────────────────
|
||||
|
||||
static Table[] makeTables(int T) {
|
||||
Table[] tables = new Table[T];
|
||||
for (int i = 0; i < T; i++) tables[i] = new Table("tbl_" + i);
|
||||
return tables;
|
||||
}
|
||||
|
||||
static Map<ColRef, Table> makeColRefMap(int N, Table[] tables) {
|
||||
Map<ColRef, Table> m = new LinkedHashMap<>();
|
||||
for (int i = 0; i < N; i++) {
|
||||
m.put(new ColRef(i), tables[i % tables.length]);
|
||||
}
|
||||
return m;
|
||||
}
|
||||
|
||||
// ── Assertions ───────────────────────────────────────────────────────────
|
||||
|
||||
static void assertEquals(Object expected, Object actual, String msg) {
|
||||
if (!expected.equals(actual))
|
||||
throw new AssertionError(msg + ": expected=" + expected + " actual=" + actual);
|
||||
}
|
||||
|
||||
static void assertTrue(boolean cond, String msg) {
|
||||
if (!cond) throw new AssertionError(msg);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
System.out.println("=== starrocks-0001: MvTableRelationAlgorithm ===");
|
||||
|
||||
DefectiveAlgorithm defAlg = new DefectiveAlgorithm();
|
||||
FixedAlgorithm fixAlg = new FixedAlgorithm();
|
||||
|
||||
// Small correctness test
|
||||
{
|
||||
int N = 50, T = 5;
|
||||
Table[] tables = makeTables(T);
|
||||
Map<ColRef, Table> colMap = makeColRefMap(N, tables);
|
||||
Set<ColRef> valid = new HashSet<>(colMap.keySet());
|
||||
List<Table> tableList = Arrays.asList(tables).subList(0, 3);
|
||||
|
||||
Result dr = defAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList));
|
||||
Result fr = fixAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList));
|
||||
|
||||
assertEquals(dr.tableToRelationId, fr.tableToRelationId, "tableToRelationId contents");
|
||||
System.out.println("PASS correctness (N=50, T=5)");
|
||||
}
|
||||
|
||||
// Benchmark
|
||||
{
|
||||
int N = 10000, T = 200;
|
||||
Table[] tables = makeTables(T);
|
||||
Map<ColRef, Table> colMap = makeColRefMap(N, tables);
|
||||
Set<ColRef> valid = new HashSet<>(colMap.keySet());
|
||||
// tableList covers only first 100 tables — forces full scan of list per entry
|
||||
List<Table> tableList = Arrays.asList(tables).subList(0, 100);
|
||||
|
||||
// warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList));
|
||||
fixAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList));
|
||||
}
|
||||
|
||||
long defNs = 0, fixNs = 0;
|
||||
int reps = 30;
|
||||
for (int i = 0; i < reps; i++) {
|
||||
defNs += defAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList)).ns;
|
||||
fixNs += fixAlg.getTableToRelationid(colMap, valid, new ArrayList<>(tableList)).ns;
|
||||
}
|
||||
defNs /= reps; fixNs /= reps;
|
||||
double ratio = (double) defNs / Math.max(1, fixNs);
|
||||
|
||||
System.out.printf("BENCH N=%d T=%d (list size=%d) reps=%d%n", N, T, tableList.size(), reps);
|
||||
System.out.printf(" defective avg: %,d ns%n", defNs);
|
||||
System.out.printf(" fixed avg: %,d ns%n", fixNs);
|
||||
System.out.printf(" speedup: %.1fx%n", ratio);
|
||||
|
||||
assertTrue(ratio >= 2.0, "Expected fixed >= 2x faster, got " + ratio + "x");
|
||||
System.out.println("PASS speedup >= 2x");
|
||||
}
|
||||
|
||||
System.out.println("=== ALL PASS ===");
|
||||
}
|
||||
}
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
BIN
defects/starrocks/unit/unit/MvTableRelationAlgorithm$Table.class
Normal file
BIN
defects/starrocks/unit/unit/MvTableRelationAlgorithm$Table.class
Normal file
Binary file not shown.
BIN
defects/starrocks/unit/unit/MvTableRelationAlgorithm.class
Normal file
BIN
defects/starrocks/unit/unit/MvTableRelationAlgorithm.class
Normal file
Binary file not shown.
|
|
@ -0,0 +1,113 @@
|
|||
# trino-0001: PushDownDereferenceThroughJoin — List.contains in stream filter → O(N²)
|
||||
|
||||
## Classification
|
||||
- **Severity**: HIGH
|
||||
- **CWE**: CWE-407 (Algorithmic Complexity — Inefficient Algorithmic Complexity)
|
||||
- **Component**: `core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushDownDereferenceThroughJoin.java`
|
||||
- **Method**: `apply()`
|
||||
|
||||
## Defect
|
||||
|
||||
`PlanNode.getOutputSymbols()` returns `List<Symbol>`. In `apply()`, this list is searched with
|
||||
`.contains()` (O(N)) in two distinct hot loops:
|
||||
|
||||
**Loop 1** (lines 127–139): `dereferenceAssignments.entrySet().forEach(entry -> { ... if (joinNode.getLeft().getOutputSymbols().contains(baseSymbol)) ... })`.
|
||||
Each iteration re-invokes `getOutputSymbols()` returning the same List and calls `.contains()` on it — O(D × S) where D = dereference count, S = output symbol count.
|
||||
|
||||
**Loop 2** (lines 152–158): `referredSymbolsInAssignments.stream().filter(symbol -> leftNode.getOutputSymbols().contains(symbol))` and the mirrored right-side filter.
|
||||
For each of R referred symbols, `.contains()` scans up to S symbols — O(R × S).
|
||||
|
||||
Both sides of the join are affected. In a wide JOIN with many projected columns and deep dereference chains (common in analytical queries over denormalized schemas), this rule runs repeatedly inside the iterative optimizer's rule-application loop, multiplying the cost.
|
||||
|
||||
**Affected lines**:
|
||||
- Line 130: `joinNode.getLeft().getOutputSymbols().contains(baseSymbol)` — `List.contains` O(S)
|
||||
- Line 133: `joinNode.getRight().getOutputSymbols().contains(baseSymbol)` — `List.contains` O(S)
|
||||
- Line 153: `.filter(symbol -> leftNode.getOutputSymbols().contains(symbol))` — `List.contains` O(S) per symbol
|
||||
- Line 157: `.filter(symbol -> rightNode.getOutputSymbols().contains(symbol))` — `List.contains` O(S) per symbol
|
||||
|
||||
## Root Cause
|
||||
|
||||
`PlanNode.getOutputSymbols()` is declared to return `List<Symbol>` (abstract method in
|
||||
`PlanNode.java`). All concrete plan nodes return `ImmutableList<Symbol>`. The caller uses `.contains()`
|
||||
for membership tests without first converting to a `Set`.
|
||||
|
||||
## Fix
|
||||
|
||||
Snapshot both output symbol lists as `ImmutableSet` before the loops. Each `ImmutableSet.copyOf()`
|
||||
is O(S), paid once. All subsequent `.contains()` calls are O(1), reducing both loops to O(D) and O(R).
|
||||
|
||||
```java
|
||||
// Before apply():
|
||||
// joinNode.getLeft().getOutputSymbols().contains(baseSymbol) — O(S) per call
|
||||
// joinNode.getRight().getOutputSymbols().contains(baseSymbol) — O(S) per call
|
||||
// leftNode.getOutputSymbols().contains(symbol) — O(S) per call
|
||||
// rightNode.getOutputSymbols().contains(symbol) — O(S) per call
|
||||
|
||||
// After (snapshot sets once before the loops):
|
||||
Set<Symbol> leftSymbols = ImmutableSet.copyOf(joinNode.getLeft().getOutputSymbols());
|
||||
Set<Symbol> rightSymbols = ImmutableSet.copyOf(joinNode.getRight().getOutputSymbols());
|
||||
// … use leftSymbols.contains / rightSymbols.contains in the forEach …
|
||||
|
||||
Set<Symbol> leftNodeSymbols = ImmutableSet.copyOf(leftNode.getOutputSymbols());
|
||||
Set<Symbol> rightNodeSymbols = ImmutableSet.copyOf(rightNode.getOutputSymbols());
|
||||
// … use leftNodeSymbols.contains / rightNodeSymbols.contains in the stream filters …
|
||||
```
|
||||
|
||||
## Complexity
|
||||
|
||||
| Before | After |
|
||||
|--------|-------|
|
||||
| O(D×S + R×S) per rule invocation | O(S + D + R) per rule invocation |
|
||||
|
||||
With D=50 dereferences, R=80 referred symbols, S=200 output symbols per side:
|
||||
Before: ~26 000 list-scans. After: 400 operations. **65× reduction**.
|
||||
|
||||
## Patch
|
||||
|
||||
```diff
|
||||
--- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushDownDereferenceThroughJoin.java
|
||||
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushDownDereferenceThroughJoin.java
|
||||
@@ -120,9 +120,12 @@ public class PushDownDereferenceThroughJoin
|
||||
Assignments.Builder leftAssignmentsBuilder = Assignments.builder();
|
||||
Assignments.Builder rightAssignmentsBuilder = Assignments.builder();
|
||||
|
||||
+ // Snapshot output symbol sets O(S) once — avoids O(S) List.contains per entry
|
||||
+ Set<Symbol> leftOutputSet = ImmutableSet.copyOf(joinNode.getLeft().getOutputSymbols());
|
||||
+ Set<Symbol> rightOutputSet = ImmutableSet.copyOf(joinNode.getRight().getOutputSymbols());
|
||||
+
|
||||
// Separate dereferences coming from left and right nodes
|
||||
dereferenceAssignments.entrySet()
|
||||
.forEach(entry -> {
|
||||
Symbol baseSymbol = getOnlyElement(extractAll(entry.getValue()));
|
||||
- if (joinNode.getLeft().getOutputSymbols().contains(baseSymbol)) {
|
||||
+ if (leftOutputSet.contains(baseSymbol)) {
|
||||
leftAssignmentsBuilder.put(entry.getKey(), entry.getValue());
|
||||
}
|
||||
- else if (joinNode.getRight().getOutputSymbols().contains(baseSymbol)) {
|
||||
+ else if (rightOutputSet.contains(baseSymbol)) {
|
||||
rightAssignmentsBuilder.put(entry.getKey(), entry.getValue());
|
||||
}
|
||||
|
||||
@@ -140,10 +144,13 @@ public class PushDownDereferenceThroughJoin
|
||||
PlanNode leftNode = createProjectNodeIfRequired(joinNode.getLeft(), leftAssignments, context.getIdAllocator());
|
||||
PlanNode rightNode = createProjectNodeIfRequired(joinNode.getRight(), rightAssignments, context.getIdAllocator());
|
||||
|
||||
+ // Snapshot post-project output sets O(S) once
|
||||
+ Set<Symbol> leftNodeSet = ImmutableSet.copyOf(leftNode.getOutputSymbols());
|
||||
+ Set<Symbol> rightNodeSet = ImmutableSet.copyOf(rightNode.getOutputSymbols());
|
||||
+
|
||||
// Prepare new output symbols for join node
|
||||
List<Symbol> referredSymbolsInAssignments = newAssignments.expressions().stream()
|
||||
.flatMap(expression -> extractAll(expression).stream())
|
||||
.collect(toList());
|
||||
|
||||
List<Symbol> newLeftOutputSymbols = referredSymbolsInAssignments.stream()
|
||||
- .filter(symbol -> leftNode.getOutputSymbols().contains(symbol))
|
||||
+ .filter(leftNodeSet::contains)
|
||||
.collect(toList());
|
||||
|
||||
List<Symbol> newRightOutputSymbols = referredSymbolsInAssignments.stream()
|
||||
- .filter(symbol -> rightNode.getOutputSymbols().contains(symbol))
|
||||
+ .filter(rightNodeSet::contains)
|
||||
.collect(toList());
|
||||
```
|
||||
208
defects/trino/unit/PushDownDerefJoinAlgorithm.java
Normal file
208
defects/trino/unit/PushDownDerefJoinAlgorithm.java
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.*;
|
||||
|
||||
/**
|
||||
* Standalone unit test for trino-0001:
|
||||
* PushDownDereferenceThroughJoin — List.contains() in stream filter → O(N²).
|
||||
*
|
||||
* Simulates the two hot loops in PushDownDereferenceThroughJoin.apply():
|
||||
* Loop 1: forEach over dereferenceAssignments, calling outputSymbols.contains()
|
||||
* Loop 2: stream.filter(s -> outputSymbols.contains(s))
|
||||
*
|
||||
* Compile: javac -d . PushDownDerefJoinAlgorithm.java
|
||||
* Run: java unit.PushDownDerefJoinAlgorithm
|
||||
*/
|
||||
public class PushDownDerefJoinAlgorithm {
|
||||
|
||||
// ── Node ─────────────────────────────────────────────────────────────────
|
||||
|
||||
static class Symbol {
|
||||
final String name;
|
||||
Symbol(String name) { this.name = name; }
|
||||
|
||||
@Override public boolean equals(Object o) {
|
||||
return o instanceof Symbol && name.equals(((Symbol) o).name);
|
||||
}
|
||||
@Override public int hashCode() { return name.hashCode(); }
|
||||
@Override public String toString() { return name; }
|
||||
}
|
||||
|
||||
// ── Defective: uses List.contains() O(N) per look-up ────────────────────
|
||||
|
||||
static class DefectivePushDown {
|
||||
List<Symbol> filterSymbols(
|
||||
List<Symbol> referred,
|
||||
List<Symbol> nodeOutputSymbols) { // List → O(N) .contains
|
||||
// Mirrors: referredSymbols.stream().filter(s -> nodeOutput.contains(s))
|
||||
return referred.stream()
|
||||
.filter(s -> nodeOutputSymbols.contains(s)) // CWE-407 site
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
Map<Symbol, Symbol> classifyDerefs(
|
||||
List<Symbol> derefs,
|
||||
List<Symbol> leftOutput, // List → O(N) .contains
|
||||
List<Symbol> rightOutput) { // List → O(N) .contains
|
||||
Map<Symbol, Symbol> result = new LinkedHashMap<>();
|
||||
for (Symbol d : derefs) {
|
||||
if (leftOutput.contains(d)) { // CWE-407 site
|
||||
result.put(d, new Symbol("left:" + d.name));
|
||||
} else if (rightOutput.contains(d)) { // CWE-407 site
|
||||
result.put(d, new Symbol("right:" + d.name));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Fixed: converts to Set once, then O(1) per look-up ──────────────────
|
||||
|
||||
static class FixedPushDown {
|
||||
List<Symbol> filterSymbols(
|
||||
List<Symbol> referred,
|
||||
List<Symbol> nodeOutputSymbols) {
|
||||
Set<Symbol> outputSet = new HashSet<>(nodeOutputSymbols); // O(N) once
|
||||
return referred.stream()
|
||||
.filter(outputSet::contains) // O(1) each
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
Map<Symbol, Symbol> classifyDerefs(
|
||||
List<Symbol> derefs,
|
||||
List<Symbol> leftOutput,
|
||||
List<Symbol> rightOutput) {
|
||||
Set<Symbol> leftSet = new HashSet<>(leftOutput); // O(N) once
|
||||
Set<Symbol> rightSet = new HashSet<>(rightOutput); // O(N) once
|
||||
Map<Symbol, Symbol> result = new LinkedHashMap<>();
|
||||
for (Symbol d : derefs) {
|
||||
if (leftSet.contains(d)) { // O(1)
|
||||
result.put(d, new Symbol("left:" + d.name));
|
||||
} else if (rightSet.contains(d)) { // O(1)
|
||||
result.put(d, new Symbol("right:" + d.name));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Result container ─────────────────────────────────────────────────────
|
||||
|
||||
static class Result {
|
||||
final String label;
|
||||
final List<Symbol> filteredLeft;
|
||||
final List<Symbol> filteredRight;
|
||||
final Map<Symbol, Symbol> classified;
|
||||
final long ns;
|
||||
|
||||
Result(String label, List<Symbol> fl, List<Symbol> fr, Map<Symbol, Symbol> cl, long ns) {
|
||||
this.label = label; this.filteredLeft = fl; this.filteredRight = fr;
|
||||
this.classified = cl; this.ns = ns;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Test driver ──────────────────────────────────────────────────────────
|
||||
|
||||
static List<Symbol> symbols(int n, String prefix) {
|
||||
List<Symbol> out = new ArrayList<>(n);
|
||||
for (int i = 0; i < n; i++) out.add(new Symbol(prefix + i));
|
||||
return out;
|
||||
}
|
||||
|
||||
static Result runDefective(int N, int D) {
|
||||
List<Symbol> leftOutput = symbols(N, "lout");
|
||||
List<Symbol> rightOutput = symbols(N, "rout");
|
||||
// derefs: first half from left, second from right
|
||||
List<Symbol> derefs = new ArrayList<>();
|
||||
for (int i = 0; i < D / 2; i++) derefs.add(leftOutput.get(i % N));
|
||||
for (int i = 0; i < D / 2; i++) derefs.add(rightOutput.get(i % N));
|
||||
|
||||
List<Symbol> referred = new ArrayList<>();
|
||||
referred.addAll(leftOutput.subList(0, Math.min(N / 2, N)));
|
||||
referred.addAll(rightOutput.subList(0, Math.min(N / 2, N)));
|
||||
|
||||
DefectivePushDown alg = new DefectivePushDown();
|
||||
long t0 = System.nanoTime();
|
||||
Map<Symbol, Symbol> classified = alg.classifyDerefs(derefs, leftOutput, rightOutput);
|
||||
List<Symbol> fl = alg.filterSymbols(referred, leftOutput);
|
||||
List<Symbol> fr = alg.filterSymbols(referred, rightOutput);
|
||||
long ns = System.nanoTime() - t0;
|
||||
return new Result("DEFECTIVE", fl, fr, classified, ns);
|
||||
}
|
||||
|
||||
static Result runFixed(int N, int D) {
|
||||
List<Symbol> leftOutput = symbols(N, "lout");
|
||||
List<Symbol> rightOutput = symbols(N, "rout");
|
||||
List<Symbol> derefs = new ArrayList<>();
|
||||
for (int i = 0; i < D / 2; i++) derefs.add(leftOutput.get(i % N));
|
||||
for (int i = 0; i < D / 2; i++) derefs.add(rightOutput.get(i % N));
|
||||
|
||||
List<Symbol> referred = new ArrayList<>();
|
||||
referred.addAll(leftOutput.subList(0, Math.min(N / 2, N)));
|
||||
referred.addAll(rightOutput.subList(0, Math.min(N / 2, N)));
|
||||
|
||||
FixedPushDown alg = new FixedPushDown();
|
||||
long t0 = System.nanoTime();
|
||||
Map<Symbol, Symbol> classified = alg.classifyDerefs(derefs, leftOutput, rightOutput);
|
||||
List<Symbol> fl = alg.filterSymbols(referred, leftOutput);
|
||||
List<Symbol> fr = alg.filterSymbols(referred, rightOutput);
|
||||
long ns = System.nanoTime() - t0;
|
||||
return new Result("FIXED", fl, fr, classified, ns);
|
||||
}
|
||||
|
||||
// ── Assertions ───────────────────────────────────────────────────────────
|
||||
|
||||
static void assertEquals(Object expected, Object actual, String msg) {
|
||||
if (!expected.equals(actual)) {
|
||||
throw new AssertionError(msg + ": expected=" + expected + " actual=" + actual);
|
||||
}
|
||||
}
|
||||
|
||||
static void assertTrue(boolean cond, String msg) {
|
||||
if (!cond) throw new AssertionError(msg);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
System.out.println("=== trino-0001: PushDownDerefJoinAlgorithm ===");
|
||||
|
||||
// Small correctness test
|
||||
{
|
||||
int N = 20, D = 10;
|
||||
Result d = runDefective(N, D);
|
||||
Result f = runFixed(N, D);
|
||||
|
||||
// Both should classify the same derefs
|
||||
assertEquals(d.classified.size(), f.classified.size(), "classified size");
|
||||
assertEquals(d.filteredLeft.size(), f.filteredLeft.size(), "filteredLeft size");
|
||||
assertEquals(d.filteredRight.size(), f.filteredRight.size(), "filteredRight size");
|
||||
System.out.println("PASS correctness (N=20, D=10)");
|
||||
}
|
||||
|
||||
// Benchmark at N=500, D=200
|
||||
{
|
||||
int N = 500, D = 200;
|
||||
// warm up
|
||||
for (int i = 0; i < 5; i++) { runDefective(N, D); runFixed(N, D); }
|
||||
|
||||
long defNs = 0, fixNs = 0;
|
||||
int reps = 20;
|
||||
for (int i = 0; i < reps; i++) {
|
||||
defNs += runDefective(N, D).ns;
|
||||
fixNs += runFixed(N, D).ns;
|
||||
}
|
||||
defNs /= reps; fixNs /= reps;
|
||||
double ratio = (double) defNs / fixNs;
|
||||
|
||||
System.out.printf("BENCH N=500 D=200 reps=%d%n", reps);
|
||||
System.out.printf(" defective avg: %,d ns%n", defNs);
|
||||
System.out.printf(" fixed avg: %,d ns%n", fixNs);
|
||||
System.out.printf(" speedup: %.1fx%n", ratio);
|
||||
|
||||
assertTrue(ratio >= 2.0, "Expected fixed to be at least 2x faster, got " + ratio + "x");
|
||||
System.out.println("PASS speedup >= 2x");
|
||||
}
|
||||
|
||||
System.out.println("=== ALL PASS ===");
|
||||
}
|
||||
}
|
||||
Binary file not shown.
Binary file not shown.
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm$Result.class
Normal file
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm$Result.class
Normal file
Binary file not shown.
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm$Symbol.class
Normal file
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm$Symbol.class
Normal file
Binary file not shown.
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm.class
Normal file
BIN
defects/trino/unit/unit/PushDownDerefJoinAlgorithm.class
Normal file
Binary file not shown.
|
|
@ -0,0 +1,104 @@
|
|||
# yugabyte-0001 — GetXReplStreamsForTable: std::find on table_id list inside per-table loop
|
||||
|
||||
**Project:** YugabyteDB
|
||||
**File:** `src/yb/master/xrepl_catalog_manager.cc`
|
||||
**Function:** `GetXReplStreamsForTable` / `DropXClusterStreamsOfTables`
|
||||
**Severity:** HIGH
|
||||
**CWE:** CWE-407 (Inefficient Algorithmic Complexity)
|
||||
|
||||
## Defect
|
||||
|
||||
`GetXReplStreamsForTable` iterates over the entire `cdc_stream_map_` (M streams), and
|
||||
for each stream performs `std::find` on `ltm->table_id()` — a protobuf repeated field
|
||||
backed by a `RepeatedPtrField` which is a linear array of string IDs (T entries):
|
||||
|
||||
```cpp
|
||||
std::vector<CDCStreamInfoPtr> CatalogManager::GetXReplStreamsForTable(
|
||||
const TableId& table_id, ...) const {
|
||||
std::vector<CDCStreamInfoPtr> streams;
|
||||
for (const auto& entry : cdc_stream_map_) { // O(M) — all streams
|
||||
auto ltm = entry.second->LockForRead();
|
||||
if (!ltm->table_id().empty() &&
|
||||
(std::find(ltm->table_id().begin(), ltm->table_id().end(), table_id) // O(T)
|
||||
!= ltm->table_id().end()) && ...) {
|
||||
streams.push_back(entry.second);
|
||||
}
|
||||
}
|
||||
return streams;
|
||||
}
|
||||
```
|
||||
|
||||
`DropXClusterStreamsOfTables` calls this function inside a loop over all dropped
|
||||
table IDs (D tables):
|
||||
|
||||
```cpp
|
||||
for (const auto& tid : table_ids) { // O(D)
|
||||
auto table_streams = GetXReplStreamsForTable(tid, cdc::XCLUSTER); // O(M × T)
|
||||
streams.insert(...);
|
||||
}
|
||||
```
|
||||
|
||||
Total: **O(D × M × T)** where:
|
||||
- D = number of dropped tables (can be the full table set during schema migration)
|
||||
- M = total CDC/XCluster streams (grows with replication topology)
|
||||
- T = tables tracked per stream (grows with multi-table replication)
|
||||
|
||||
This is cubic in the worst case. At D=100 tables, M=50 streams, T=20 tables/stream,
|
||||
that is 100,000 string comparisons where 100 would suffice.
|
||||
|
||||
The same pattern appears at lines 2044–2049 (AddTableToXReplStream) and 2162
|
||||
(GetNonUserTablesInStream), compounding the impact.
|
||||
|
||||
## Fix
|
||||
|
||||
Build an inverted index from `TableId → set<CDCStreamInfoPtr>` once (on load and on
|
||||
stream creation/deletion) and perform O(1) lookups instead of O(M×T) full scans.
|
||||
|
||||
Alternatively, for the `DropXClusterStreamsOfTables` hot path, batch all table IDs
|
||||
and iterate `cdc_stream_map_` once:
|
||||
|
||||
```cpp
|
||||
Status CatalogManager::DropXClusterStreamsOfTables(
|
||||
const std::unordered_set<TableId>& table_ids) {
|
||||
if (table_ids.empty()) return Status::OK();
|
||||
|
||||
std::vector<CDCStreamInfoPtr> streams;
|
||||
{
|
||||
SharedLock lock(mutex_);
|
||||
for (const auto& [_, stream_info] : cdc_stream_map_) { // O(M) — one pass
|
||||
auto ltm = stream_info->LockForRead();
|
||||
if (ltm->is_deleting() || ltm->namespace_id().empty()) continue;
|
||||
// O(T) per stream but only one pass over M streams total
|
||||
for (const auto& tid : ltm->table_id()) {
|
||||
if (table_ids.count(tid)) { // O(1) hash lookup
|
||||
streams.push_back(stream_info);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
With this pattern: **O(M × T)** total instead of **O(D × M × T)**.
|
||||
For the long-term fix, maintain a `std::unordered_multimap<TableId, CDCStreamInfoPtr>`
|
||||
as a secondary index, giving O(D) lookup.
|
||||
|
||||
## Impact
|
||||
|
||||
During table drops in a namespace with active CDC streams, the master leader stalls
|
||||
proportionally to D×M×T. In a large multi-tenant deployment with hundreds of tables
|
||||
and dozens of replication streams each tracking tens of tables, this causes
|
||||
multi-second stalls in master processing, delaying tablet cleanup and potentially
|
||||
triggering master step-down timeouts.
|
||||
|
||||
## Location
|
||||
|
||||
```
|
||||
src/yb/master/xrepl_catalog_manager.cc
|
||||
GetXReplStreamsForTable() lines 790–814
|
||||
DropXClusterStreamsOfTables() lines 606–618 (calls GetXReplStreamsForTable in loop)
|
||||
AddTableToXReplStream() lines 2029–2050 (same pattern)
|
||||
GetNonUserTablesInStream context lines 2154–2165 (same pattern)
|
||||
```
|
||||
207
defects/yugabyte/unit/YugabyteTest.java
Normal file
207
defects/yugabyte/unit/YugabyteTest.java
Normal file
|
|
@ -0,0 +1,207 @@
|
|||
package unit;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* yugabyte-0001: GetXReplStreamsForTable std::find on table_id list inside per-table loop
|
||||
*
|
||||
* Models xrepl_catalog_manager.cc:
|
||||
* GetXReplStreamsForTable: for each stream, std::find on table_id repeated field
|
||||
* DropXClusterStreamsOfTables: calls GetXReplStreamsForTable in a loop over table_ids
|
||||
*
|
||||
* Total defective: O(D × M × T) where D=dropped tables, M=streams, T=tables per stream
|
||||
* Total fixed: O(M × T) — one pass + O(1) hash lookup
|
||||
*
|
||||
* Compile: javac -d . YugabyteTest.java
|
||||
* Run: java unit.YugabyteTest
|
||||
*/
|
||||
public class YugabyteTest {
|
||||
|
||||
static class CDCStreamInfo {
|
||||
final String streamId;
|
||||
final List<String> tableIds; // backed by protobuf RepeatedPtrField (array)
|
||||
boolean started_deleting = false;
|
||||
|
||||
CDCStreamInfo(String streamId, List<String> tableIds) {
|
||||
this.streamId = streamId;
|
||||
this.tableIds = tableIds;
|
||||
}
|
||||
}
|
||||
|
||||
// ---- DEFECTIVE: O(D × M × T) ----
|
||||
static List<String> getXReplStreamsForTable_defective(
|
||||
String tableId,
|
||||
List<CDCStreamInfo> cdcStreamMap,
|
||||
int[] comparisonCount) {
|
||||
List<String> streams = new ArrayList<>();
|
||||
for (CDCStreamInfo stream : cdcStreamMap) { // O(M)
|
||||
if (stream.started_deleting) continue;
|
||||
// std::find on table_id list — O(T) linear scan
|
||||
boolean found = false;
|
||||
for (String tid : stream.tableIds) { // O(T)
|
||||
comparisonCount[0]++;
|
||||
if (tid.equals(tableId)) { found = true; break; }
|
||||
}
|
||||
if (found) streams.add(stream.streamId);
|
||||
}
|
||||
return streams;
|
||||
}
|
||||
|
||||
static List<String> dropXClusterStreamsOfTables_defective(
|
||||
Set<String> tableIds,
|
||||
List<CDCStreamInfo> cdcStreamMap,
|
||||
int[] comparisonCount) {
|
||||
List<String> affectedStreams = new ArrayList<>();
|
||||
for (String tid : tableIds) { // O(D)
|
||||
affectedStreams.addAll(
|
||||
getXReplStreamsForTable_defective(tid, cdcStreamMap, comparisonCount));
|
||||
}
|
||||
return affectedStreams;
|
||||
}
|
||||
|
||||
// ---- FIXED: single pass O(M × T) with O(1) hash lookup ----
|
||||
static List<String> dropXClusterStreamsOfTables_fixed(
|
||||
Set<String> tableIds,
|
||||
List<CDCStreamInfo> cdcStreamMap,
|
||||
int[] comparisonCount) {
|
||||
List<String> affectedStreams = new ArrayList<>();
|
||||
|
||||
for (CDCStreamInfo stream : cdcStreamMap) { // O(M) — single pass
|
||||
if (stream.started_deleting) continue;
|
||||
for (String tid : stream.tableIds) { // O(T) per stream
|
||||
comparisonCount[0]++;
|
||||
if (tableIds.contains(tid)) { // O(1) hash set lookup
|
||||
affectedStreams.add(stream.streamId);
|
||||
break; // found one match, don't add stream twice
|
||||
}
|
||||
}
|
||||
}
|
||||
return affectedStreams;
|
||||
}
|
||||
|
||||
// Build a test CDC stream map
|
||||
static List<CDCStreamInfo> buildStreamMap(int M, int T, int totalTables) {
|
||||
List<CDCStreamInfo> map = new ArrayList<>();
|
||||
for (int i = 0; i < M; i++) {
|
||||
List<String> tables = new ArrayList<>();
|
||||
for (int j = 0; j < T; j++) {
|
||||
tables.add("table-" + ((i * T + j) % totalTables));
|
||||
}
|
||||
map.add(new CDCStreamInfo("stream-" + i, tables));
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
int pass = 0, fail = 0;
|
||||
|
||||
// -- Test 1: correctness — small scenario
|
||||
{
|
||||
List<CDCStreamInfo> streams = new ArrayList<>();
|
||||
streams.add(new CDCStreamInfo("s1", Arrays.asList("t1", "t2", "t3")));
|
||||
streams.add(new CDCStreamInfo("s2", Arrays.asList("t2", "t4")));
|
||||
streams.add(new CDCStreamInfo("s3", Arrays.asList("t5", "t6")));
|
||||
|
||||
Set<String> toDrop = new HashSet<>(Arrays.asList("t2", "t5"));
|
||||
|
||||
int[] cmpDef = {0}, cmpFix = {0};
|
||||
List<String> resultDef = dropXClusterStreamsOfTables_defective(toDrop, streams, cmpDef);
|
||||
List<String> resultFix = dropXClusterStreamsOfTables_fixed(toDrop, streams, cmpFix);
|
||||
|
||||
Collections.sort(resultDef);
|
||||
Collections.sort(resultFix);
|
||||
|
||||
if (resultDef.equals(resultFix)) {
|
||||
System.out.printf("PASS test1: correctness — both=%s%n", resultDef);
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test1: defective=%s fixed=%s%n", resultDef, resultFix);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
// -- Test 2: complexity comparison
|
||||
{
|
||||
int D = 50; // tables being dropped
|
||||
int M = 100; // CDC streams
|
||||
int T = 20; // tables per stream
|
||||
int totalTables = 200;
|
||||
|
||||
List<CDCStreamInfo> streamMap = buildStreamMap(M, T, totalTables);
|
||||
|
||||
// Drop first D tables
|
||||
Set<String> toDrop = new HashSet<>();
|
||||
for (int i = 0; i < D; i++) toDrop.add("table-" + i);
|
||||
|
||||
int[] cmpDef = {0}, cmpFix = {0};
|
||||
List<String> resultDef = dropXClusterStreamsOfTables_defective(toDrop, streamMap, cmpDef);
|
||||
List<String> resultFix = dropXClusterStreamsOfTables_fixed(toDrop, streamMap, cmpFix);
|
||||
|
||||
System.out.printf("test2: defective comparisons=%d fixed comparisons=%d (D=%d M=%d T=%d)%n",
|
||||
cmpDef[0], cmpFix[0], D, M, T);
|
||||
|
||||
// Defective adds a stream once per matching dropped table (may have duplicates);
|
||||
// fixed adds each stream at most once. Compare as sets of unique stream IDs.
|
||||
Set<String> setDef = new HashSet<>(resultDef);
|
||||
Set<String> setFix = new HashSet<>(resultFix);
|
||||
|
||||
if (setDef.equals(setFix) && cmpFix[0] < cmpDef[0]) {
|
||||
System.out.println("PASS test2: fixed is more efficient and produces same (unique) result");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test2: match=%b efficient=%b%n",
|
||||
setDef.equals(setFix), cmpFix[0] < cmpDef[0]);
|
||||
fail++;
|
||||
}
|
||||
|
||||
// Show expected complexity
|
||||
System.out.printf(" Expected defective ~O(D×M×T)=%d, actual=%d%n", D * M * T, cmpDef[0]);
|
||||
System.out.printf(" Expected fixed ~O(M×T)=%d, actual=%d%n", M * T, cmpFix[0]);
|
||||
}
|
||||
|
||||
// -- Test 3: deleted streams are excluded
|
||||
{
|
||||
CDCStreamInfo deletingStream = new CDCStreamInfo("s-deleting", Arrays.asList("t1", "t2"));
|
||||
deletingStream.started_deleting = true;
|
||||
|
||||
List<CDCStreamInfo> streams = new ArrayList<>();
|
||||
streams.add(deletingStream);
|
||||
streams.add(new CDCStreamInfo("s-active", Arrays.asList("t1")));
|
||||
|
||||
Set<String> toDrop = new HashSet<>(Collections.singletonList("t1"));
|
||||
|
||||
int[] cmpDef = {0}, cmpFix = {0};
|
||||
List<String> resultDef = dropXClusterStreamsOfTables_defective(toDrop, streams, cmpDef);
|
||||
List<String> resultFix = dropXClusterStreamsOfTables_fixed(toDrop, streams, cmpFix);
|
||||
|
||||
Collections.sort(resultDef);
|
||||
Collections.sort(resultFix);
|
||||
|
||||
if (resultDef.equals(resultFix) && resultFix.equals(Collections.singletonList("s-active"))) {
|
||||
System.out.printf("PASS test3: deleting stream excluded — result=%s%n", resultFix);
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test3: def=%s fix=%s%n", resultDef, resultFix);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
// -- Test 4: empty table_ids set
|
||||
{
|
||||
List<CDCStreamInfo> streams = buildStreamMap(10, 5, 50);
|
||||
int[] c1 = {0}, c2 = {0};
|
||||
List<String> r1 = dropXClusterStreamsOfTables_defective(new HashSet<>(), streams, c1);
|
||||
List<String> r2 = dropXClusterStreamsOfTables_fixed(new HashSet<>(), streams, c2);
|
||||
if (r1.isEmpty() && r2.isEmpty()) {
|
||||
System.out.println("PASS test4: empty input — both return empty");
|
||||
pass++;
|
||||
} else {
|
||||
System.out.printf("FAIL test4: def=%s fix=%s%n", r1, r2);
|
||||
fail++;
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf("%nResults: %d passed, %d failed%n", pass, fail);
|
||||
if (fail > 0) System.exit(1);
|
||||
}
|
||||
}
|
||||
|
|
@ -1 +1 @@
|
|||
93e63b76eed03e223cacf4b27544c527 undefect-cwe407-2026-03-27.pdf
|
||||
25e9125115c5f626aa76d76fa7c2b85a undefect-cwe407-2026-03-27.pdf
|
||||
|
|
|
|||
|
|
@ -39,8 +39,8 @@ A single well-crafted implementation serves as the genetic blueprint.
|
|||
4. **Harvest Stage:** Mature implementations compile into comprehensive documentation, ready for use
|
||||
|
||||
Code propagates according to its kind — clean architecture begets clean implementations,
|
||||
elegant solutions inspire elegant variations. The process of generating 462 validated
|
||||
defect patches across 209 ecosystems in a single research wave demonstrates how truth,
|
||||
elegant solutions inspire elegant variations. The process of generating 465 validated
|
||||
defect patches across 212 ecosystems in a single research wave demonstrates how truth,
|
||||
properly seeded, multiplies. Each tested patch validates the correctness of the original
|
||||
diagnosis & extends light into new programming paradigms.
|
||||
|
||||
|
|
@ -159,7 +159,7 @@ the missing linkages, applied them, tested them, and benchmarked them across eve
|
|||
confirmed site — compiler, routing, database, build tool, event streaming, web framework,
|
||||
query optimizer, and browser runtime.
|
||||
|
||||
**462 sites patched. 3 deferred (PostgreSQL -0001/-0005; MongoDB -0005 IndexBounds).
|
||||
**465 sites patched. 3 deferred (PostgreSQL -0001/-0005; MongoDB -0005 IndexBounds).
|
||||
1 fixable-upstream (Erlang OTP). 1 fixable-pending (swipl-0003). 2 not-worth-fixing.
|
||||
3 unpatched (Minecraft, Create mod). No language left behind.
|
||||
|
||||
|
|
@ -375,6 +375,9 @@ stacks, Spark schemas — this is the dominant build cost.
|
|||
| tidb-0006 | TiDB | `planner/core/` — `slices.Contains` in aggregate pushdown (188×) | **PATCHED** |
|
||||
| tidb-0007 | TiDB | `planner/core/` — `slices.Contains` in index merge path selection (188×) | **PATCHED** |
|
||||
| tidb-0008 | TiDB | `planner/core/` — `slices.Contains` in expression rewriter (188×) | **PATCHED** |
|
||||
| scylladb-0001 | ScyllaDB | `service/storage_proxy.cc:7135` — `std::find` on replica-set vector in `intersection()`, O(V×RF²) per range scan; fix: `unordered_set<host_id>` | **PATCHED** |
|
||||
| yugabyte-0001 | YugabyteDB | `master/xrepl_catalog_manager.cc:793` — `std::find` on protobuf `table_id` field in CDC stream loop; O(D×M×T) cubic (66×) | **PATCHED** |
|
||||
| foundationdb-0001 | FoundationDB | `DDRelocationQueue.actor.cpp:465` — `std::count` on `servers` vector in `canLaunchSrc()` double loop; O(S×R×S') | **PATCHED** |
|
||||
| kubernetes-0001 | Kubernetes | `pkg/controller/job/job_controller.go` — `slices.Contains(Values)` O(C×R×V) per failed pod in failure policy eval; fix: `HashSet` per requirement (45×) | **PATCHED** |
|
||||
| kubernetes-0002 | Kubernetes | `pkg/controller/garbagecollector/` — `slices.Contains(ownerUIDs)` O(refs×UIDs) per GC cycle; fix: `map[types.UID]struct{}` (150×) | **PATCHED** |
|
||||
| kubernetes-0003 | Kubernetes | `pkg/controller/job/job_controller.go:1357` — `hasJobTrackingFinalizer()` called again in pass 2 despite `uidsWithFinalizer` set already built in pass 1; redundant O(P×F) scan; fix: `uidsWithFinalizer.Has(pod.UID)` (1.67×) | **PATCHED** |
|
||||
|
|
@ -737,7 +740,7 @@ where D is the depth of the diamond chain. For a diamond of depth 10, that is 2^
|
|||
1,024 redundant node visits per edge check. Large modpacks produce diamond dependency
|
||||
chains with depths in this range.
|
||||
|
||||
**462 sites patched. 3 deferred (PostgreSQL -0001/-0005; MongoDB -0005 IndexBounds). 1 fixable-upstream (Erlang OTP — sltab patch). 1 fixable-pending (swipl-0003 attr_unify_hook). 2 not-worth-fixing. 3 unpatched (Minecraft, Create mod). 12 CLEAN (WireGuard-tools, Solana, git, JGit, Dask, OSRM, Buck2, DGL, Protocol Buffers, gRPC Python, Apache Beam, Apache Samza).**
|
||||
**465 sites patched. 3 deferred (PostgreSQL -0001/-0005; MongoDB -0005 IndexBounds). 1 fixable-upstream (Erlang OTP — sltab patch). 1 fixable-pending (swipl-0003 attr_unify_hook). 2 not-worth-fixing. 3 unpatched (Minecraft, Create mod). 12 CLEAN (WireGuard-tools, Solana, git, JGit, Dask, OSRM, Buck2, DGL, Protocol Buffers, gRPC Python, Apache Beam, Apache Samza).**
|
||||
|
||||
---
|
||||
|
||||
|
|
@ -1326,6 +1329,29 @@ This fires on every `updateSchema()` DDL commit. Fix: change `deletes` to `HashS
|
|||
and Apache Samza are CLEAN: Beam's pipeline graph uses `ImmutableSet`/`HashSet` throughout; Samza's
|
||||
`topologicalSort()` uses `HashSet<JobNode> visited`.
|
||||
|
||||
**ScyllaDB — scylladb-0001 (MEDIUM)**
|
||||
|
||||
`storage_proxy::intersection()` computes replica-set intersection using `std::remove_copy_if` with an
|
||||
inner `std::find` closure over the second replica list — O(|l1|×|l2|). Called twice per vnode in the
|
||||
range-merge scatter/gather read path. With 256 vnodes and RF=5, each range scan accumulates 512 O(RF²)
|
||||
intersection calls. Fix: pre-build `unordered_set<host_id>` for O(|l1|+|l2|). Affects the legacy
|
||||
vnode path only (tablet clusters bypass via runtime guard).
|
||||
|
||||
**YugabyteDB — yugabyte-0001 (HIGH, 66×)**
|
||||
|
||||
`GetXReplStreamsForTable()` in the xCluster/CDC catalog manager iterates all M CDC streams and calls
|
||||
`std::find` on the protobuf `table_id` repeated field (T entries) per stream, in a per-dropped-table
|
||||
loop — O(D×M×T) cubic. At D=50, M=100, T=20: 94,750 comparisons vs 1,430 for the fix. The same
|
||||
pattern appears in `AddTableToXReplStream` and `GetNonUserTablesInStream`. Fix: single pass with
|
||||
`unordered_set` on the dropped-table set.
|
||||
|
||||
**FoundationDB — foundationdb-0001 (MEDIUM)**
|
||||
|
||||
`canLaunchSrc()` in `DDRelocationQueue` checks source-server load by iterating `relocation.src` (S)
|
||||
servers and for each scanning `cancellableRelocations` (R entries) with `std::count` on each
|
||||
relocation's source server list — O(S×R×S'). Called in the hot relocation-dispatch loop. Fix:
|
||||
pre-build `unordered_map<UID, vector<int>>` from server UID to cancellable relocation indices.
|
||||
|
||||
**CFEngine** — **cfe-0001/0002/0003 PATCHED.** `getindices()`, `unique()`, and
|
||||
`maparray()` all used `RlistAppendScalarIdemp()` — which calls `RlistKeyIn()`, an O(N)
|
||||
linked-list walk — as a dedup primitive. `unique()` is a first-class CFEngine policy
|
||||
|
|
|
|||
Binary file not shown.
Loading…
Add table
Add a link
Reference in a new issue