engineering/java-topology
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

whitepaper: re-add 10 missing entries + 11 new defects this session, count 578→590; rebuild PDF

Browse source
This commit is contained in:
russell@unturf.com 2026-03-27 22:18:31 -04:00
parent e4ee168b1e
commit 2e4f7807d5
401 changed files with 3914 additions and 114 deletions
Download patch file Download diff file Expand all files Collapse all files

1
defects/flink/patch/flink-0001.patch
View file

@ -1,3 +1,4 @@
# UNDF: UNDF-2026-000000072
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -117,8 +117,8 @@

99
defects/flink/patch/flink-0005-dynamicpartitionpruning-fieldnames-indexof-hashmap.md Normal file
View file

@ -0,0 +1,99 @@
# flink-0005: DynamicPartitionPruningUtils — List.indexOf + List.contains O(A×F + K×A) → O(F + K)
## Metadata
- **Project**: Apache Flink
- **Component**: `flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java`
- **CWE**: CWE-407 (Inefficient Algorithmic Complexity)
- **Severity**: MEDIUM
- **Method**: `convertDppFactSide()` lines 325–334
- **Hot path**: Query planning for every batch job using dynamic partition pruning (star-schema joins)
## Location
```
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java
method: convertDppFactSide() lines 325–334
```
## Defective code
```java
// O(A×F): indexOf on List<String> called A times — A = accepted filter fields, F = table columns
List<Integer> acceptedFieldIndices =
acceptedFilterFields.stream()
.map(f -> scan.getRowType().getFieldNames().indexOf(f)) // O(F) per field
.collect(Collectors.toList());
// O(K×A): List.contains() called K times — K = join keys, A = accepted field indices
List<Integer> dynamicFilteringFieldIndices = new ArrayList<>();
for (int i = 0; i < joinKeys.size(); ++i) {
if (acceptedFieldIndices.contains(joinKeys.get(i))) { // O(A) linear scan
dynamicFilteringFieldIndices.add(dimSideJoinKey.get(i));
}
}
```
### Why this is O(A×F + K×A)
**Part 1 — `indexOf` loop (line 327):**
`scan.getRowType().getFieldNames()` returns a `List<String>` of F field names. `.indexOf(f)` scans the list linearly → O(F) per call. Called once per accepted filter field (A calls) → total O(A×F).
**Part 2 — `contains` loop (line 331):**
`acceptedFieldIndices` is a `List<Integer>`. `.contains(joinKeys.get(i))` scans the list linearly → O(A) per call. Called once per join key (K calls) → total O(K×A).
For a wide partitioned table (F=200 columns, A=20 accepted fields, K=20 join keys):
- Part 1: 20 × 200 = 4,000 string comparisons
- Part 2: 20 × 20 = 400 integer comparisons
- **Total: 4,400 ops vs O(F + K) = 220 ops → ~20× overhead**
At F=500, A=50, K=50: **27,500 ops vs 550 ops → 50× overhead**.
## Fix
Build a `Map<String, Integer>` from field name → index once in O(F), then do O(1) lookups. Replace `acceptedFieldIndices` list with a `Set<Integer>` for O(1) membership tests.
```java
// O(F): build name→index map once
List<String> fieldNames = scan.getRowType().getFieldNames();
Map<String, Integer> fieldNameToIndex = new HashMap<>(fieldNames.size() * 2);
for (int i = 0; i < fieldNames.size(); i++) {
fieldNameToIndex.put(fieldNames.get(i), i);
}
// O(A): O(1) map lookup per field
Set<Integer> acceptedFieldIndexSet = new HashSet<>();
List<Integer> acceptedFieldIndices = new ArrayList<>();
for (String f : acceptedFilterFields) {
Integer idx = fieldNameToIndex.get(f); // O(1)
if (idx != null) {
acceptedFieldIndices.add(idx);
acceptedFieldIndexSet.add(idx);
}
}
// O(K): O(1) set lookup per join key
List<Integer> dynamicFilteringFieldIndices = new ArrayList<>();
for (int i = 0; i < joinKeys.size(); ++i) {
if (acceptedFieldIndexSet.contains(joinKeys.get(i))) { // O(1)
dynamicFilteringFieldIndices.add(dimSideJoinKey.get(i));
}
}
```
## Complexity analysis
| Scenario | Before (Part 1 + Part 2) | After |
|----------|--------------------------|-------|
| F=50, A=10, K=10 | 500 + 100 = 600 ops | 50 + 10 + 10 = 70 ops — 8.6× |
| F=100, A=20, K=20 | 2,000 + 400 = 2,400 ops | 100 + 20 + 20 = 140 ops — 17× |
| F=200, A=30, K=30 | 6,000 + 900 = 6,900 ops | 200 + 30 + 30 = 260 ops — 26.5× |
| F=500, A=50, K=50 | 25,000 + 2,500 = 27,500 ops | 500 + 50 + 50 = 600 ops — 45.8× |
## Notes
Dynamic partition pruning (DPP) is used in batch Flink jobs for star-schema queries
(e.g., TPC-DS queries). `convertDppFactSide` is called once per fact table scan during
query planning. Wide tables (typical in data warehouse workloads) with many partition
columns experience the most overhead.
This defect was present alongside a similar pattern: `acceptedFilterFields.stream().map(f -> getFieldNames().indexOf(f))` — both the building and querying phases are linear-scan based.

326
defects/flink/unit/FlinkDynamicPartitionPruningTest.java Normal file
View file

@ -0,0 +1,326 @@
package unit;
import java.util.*;
/**
* CWE-407 unit test — flink-0005
*
* DynamicPartitionPruningUtils.convertDppFactSide() uses:
* 1. List<String>.indexOf(f) inside a stream.map() — O(A×F)
* 2. List<Integer>.contains() inside a for loop — O(K×A)
*
* where A = accepted filter fields, F = total table columns, K = join keys.
*
* Fix: build a Map<String,Integer> once in O(F), use Set<Integer> for membership in O(1).
*
* No JUnit. Run:
* javac -d . FlinkDynamicPartitionPruningTest.java
* java -ea unit.FlinkDynamicPartitionPruningTest
*/
public class FlinkDynamicPartitionPruningTest {
// ---------------------------------------------------------------------------
// SLOW path: simulates defective DynamicPartitionPruningUtils.convertDppFactSide()
//
// acceptedFieldIndices = acceptedFilterFields.stream()
// .map(f -> fieldNames.indexOf(f)) // O(F) per field
// .collect(toList());
//
// for (int i = 0; i < joinKeys.size(); ++i) {
// if (acceptedFieldIndices.contains(joinKeys.get(i))) // O(A) per key
// result.add(dimSideJoinKey.get(i));
// }
//
// Returns: (result indices, op count)
// ---------------------------------------------------------------------------
static long[] slowConvertDpp(
List<String> fieldNames,
List<String> acceptedFilterFields,
List<Integer> joinKeys,
List<Integer> dimSideJoinKey) {
long ops = 0;
// Part 1: indexOf — O(A × F)
List<Integer> acceptedFieldIndices = new ArrayList<>();
for (String f : acceptedFilterFields) {
for (int i = 0; i < fieldNames.size(); i++) {
ops++;
if (fieldNames.get(i).equals(f)) {
acceptedFieldIndices.add(i);
break;
}
}
}
// Part 2: List.contains — O(K × A)
List<Integer> result = new ArrayList<>();
for (int i = 0; i < joinKeys.size(); ++i) {
int key = joinKeys.get(i);
for (int accepted : acceptedFieldIndices) {
ops++;
if (accepted == key) {
result.add(dimSideJoinKey.get(i));
break;
}
}
}
return new long[]{result.size(), ops};
}
// ---------------------------------------------------------------------------
// FAST path: Map<String,Integer> + Set<Integer>
//
// Map<String,Integer> nameToIdx = build once from fieldNames — O(F)
// Set<Integer> acceptedSet = acceptedFilterFields.stream()
// .map(nameToIdx::get) — O(A)
// .collect(toSet())
//
// for (int i = 0; i < joinKeys.size(); ++i) {
// if (acceptedSet.contains(joinKeys.get(i))) — O(1)
// result.add(dimSideJoinKey.get(i));
// }
//
// Returns: (result indices, op count)
// ---------------------------------------------------------------------------
static long[] fastConvertDpp(
List<String> fieldNames,
List<String> acceptedFilterFields,
List<Integer> joinKeys,
List<Integer> dimSideJoinKey) {
long ops = 0;
// Build name→index map: O(F)
Map<String, Integer> nameToIdx = new HashMap<>(fieldNames.size() * 2);
for (int i = 0; i < fieldNames.size(); i++) {
nameToIdx.put(fieldNames.get(i), i);
ops++; // one write per field
}
// Build accepted set: O(A)
Set<Integer> acceptedSet = new HashSet<>();
List<Integer> acceptedFieldIndices = new ArrayList<>();
for (String f : acceptedFilterFields) {
Integer idx = nameToIdx.get(f); // O(1)
ops++;
if (idx != null) {
acceptedSet.add(idx);
acceptedFieldIndices.add(idx);
}
}
// Membership test: O(K)
List<Integer> result = new ArrayList<>();
for (int i = 0; i < joinKeys.size(); ++i) {
ops++; // O(1) set lookup
if (acceptedSet.contains(joinKeys.get(i))) {
result.add(dimSideJoinKey.get(i));
}
}
return new long[]{result.size(), ops};
}
// ---------------------------------------------------------------------------
// Build test data
// ---------------------------------------------------------------------------
/** Generate fieldNames = ["col_0", "col_1", ..., "col_{F-1}"] */
static List<String> buildFieldNames(int F) {
List<String> names = new ArrayList<>(F);
for (int i = 0; i < F; i++) names.add("col_" + i);
return names;
}
/**
* Select A evenly-spaced field names from fieldNames as the accepted filter fields.
* Also build joinKeys as indices into fieldNames (same set), and dimSideJoinKey as identity.
*/
static Object[] buildScenario(int F, int A, int K) {
List<String> fieldNames = buildFieldNames(F);
// acceptedFilterFields: A evenly-spaced field names
List<String> accepted = new ArrayList<>(A);
for (int i = 0; i < A; i++) {
int idx = (int) ((long) i * F / A);
accepted.add(fieldNames.get(idx));
}
// joinKeys: K evenly-spaced indices into fieldNames (subset overlapping with accepted)
List<Integer> joinKeys = new ArrayList<>(K);
List<Integer> dimSideJoinKey = new ArrayList<>(K);
for (int i = 0; i < K; i++) {
int idx = (int) ((long) i * F / K);
joinKeys.add(idx);
dimSideJoinKey.add(i * 10); // arbitrary dim-side key
}
return new Object[]{fieldNames, accepted, joinKeys, dimSideJoinKey};
}
// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------
static void test(String name, boolean cond) {
if (!cond) throw new AssertionError("FAIL: " + name);
System.out.println("PASS: " + name);
}
// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------
@SuppressWarnings("unchecked")
public static void main(String[] args) {
System.out.println("=== flink-0005: DynamicPartitionPruningUtils indexOf+contains O(A*F+K*A) ===");
System.out.println();
// T1: correctness — small scenario F=20, A=5, K=5
{
Object[] s = buildScenario(20, 5, 5);
List<String> fn = (List<String>) s[0];
List<String> af = (List<String>) s[1];
List<Integer> jk = (List<Integer>) s[2];
List<Integer> dk = (List<Integer>) s[3];
long[] slow = slowConvertDpp(fn, af, jk, dk);
long[] fast = fastConvertDpp(fn, af, jk, dk);
test("T1: slow and fast return same result count (F=20,A=5,K=5)",
slow[0] == fast[0]);
System.out.printf("T1: result=%d slow-ops=%d fast-ops=%d%n",
slow[0], slow[1], fast[1]);
}
// T2: correctness — larger scenario F=100, A=20, K=20
{
Object[] s = buildScenario(100, 20, 20);
List<String> fn = (List<String>) s[0];
List<String> af = (List<String>) s[1];
List<Integer> jk = (List<Integer>) s[2];
List<Integer> dk = (List<Integer>) s[3];
long[] slow = slowConvertDpp(fn, af, jk, dk);
long[] fast = fastConvertDpp(fn, af, jk, dk);
test("T2: slow and fast return same result count (F=100,A=20,K=20)",
slow[0] == fast[0]);
}
// T3: op count comparison — F=100, A=30, K=30
// slow: O(A*F + K*A) = 30*100 + 30*30 = 3000 + 900 = 3900
// fast: O(F + A + K) = 100 + 30 + 30 = 160
{
int F = 100, A = 30, K = 30;
Object[] s = buildScenario(F, A, K);
List<String> fn = (List<String>) s[0];
List<String> af = (List<String>) s[1];
List<Integer> jk = (List<Integer>) s[2];
List<Integer> dk = (List<Integer>) s[3];
long[] slow = slowConvertDpp(fn, af, jk, dk);
long[] fast = fastConvertDpp(fn, af, jk, dk);
System.out.printf("T3: F=%d A=%d K=%d — slow-ops=%d fast-ops=%d ratio=%.1fx%n",
F, A, K, slow[1], fast[1], (double) slow[1] / fast[1]);
test("T3: slow ops > fast ops (F=100, A=30, K=30)", slow[1] > fast[1]);
test("T3: slow ops >= A*F/2 (lower bound for O(A*F))", slow[1] >= (long) A * F / 2);
test("T3: fast ops <= F + A + K + 10 (linear bound)", fast[1] <= F + A + K + 10);
double ratio = (double) slow[1] / fast[1];
test("T3: speedup >= 5x at F=100,A=30,K=30", ratio >= 5.0);
}
// T4: scaling — doubling F should ~2x slow Part 1, ~1x fast
{
int A = 20, K = 20;
int F1 = 100, F2 = 200;
Object[] s1 = buildScenario(F1, A, K);
Object[] s2 = buildScenario(F2, A, K);
long[] slowF1 = slowConvertDpp(
(List<String>) s1[0], (List<String>) s1[1],
(List<Integer>) s1[2], (List<Integer>) s1[3]);
long[] slowF2 = slowConvertDpp(
(List<String>) s2[0], (List<String>) s2[1],
(List<Integer>) s2[2], (List<Integer>) s2[3]);
long[] fastF1 = fastConvertDpp(
(List<String>) s1[0], (List<String>) s1[1],
(List<Integer>) s1[2], (List<Integer>) s1[3]);
long[] fastF2 = fastConvertDpp(
(List<String>) s2[0], (List<String>) s2[1],
(List<Integer>) s2[2], (List<Integer>) s2[3]);
double slowRatio = (double) slowF2[1] / slowF1[1];
double fastRatio = (double) fastF2[1] / fastF1[1];
System.out.printf("T4: F=%d slow=%d fast=%d%n", F1, slowF1[1], fastF1[1]);
System.out.printf("T4: F=%d slow=%d fast=%d%n", F2, slowF2[1], fastF2[1]);
System.out.printf("T4: slow ratio=%.2f (expect ~2.0 for O(F)), fast ratio=%.2f%n",
slowRatio, fastRatio);
test("T4: slow ops grow with F (ratio >= 1.5)", slowRatio >= 1.5);
test("T4: fast ops grow with F but much slower (ratio <= slowRatio)",
fastRatio <= slowRatio);
}
// T5: large scenario — measure speedup at F=500, A=50, K=50
{
int F = 500, A = 50, K = 50;
Object[] s = buildScenario(F, A, K);
List<String> fn = (List<String>) s[0];
List<String> af = (List<String>) s[1];
List<Integer> jk = (List<Integer>) s[2];
List<Integer> dk = (List<Integer>) s[3];
long[] slow = slowConvertDpp(fn, af, jk, dk);
long[] fast = fastConvertDpp(fn, af, jk, dk);
double ratio = (double) slow[1] / fast[1];
System.out.printf("T5: F=%d A=%d K=%d — slow-ops=%d fast-ops=%d speedup=%.1fx%n",
F, A, K, slow[1], fast[1], ratio);
test("T5: slow ops >= A*F/2 (O(A*F) lower bound)", slow[1] >= (long) A * F / 2);
test("T5: fast ops <= F + A*2 + K*2 (O(F+A+K) bound)",
fast[1] <= F + A * 2 + K * 2);
test("T5: speedup >= 10x at F=500,A=50,K=50", ratio >= 10.0);
test("T5: results match", slow[0] == fast[0]);
}
// T6: wall-clock at F=1000, A=100, K=100
{
int F = 1000, A = 100, K = 100;
Object[] s = buildScenario(F, A, K);
List<String> fn = (List<String>) s[0];
List<String> af = (List<String>) s[1];
List<Integer> jk = (List<Integer>) s[2];
List<Integer> dk = (List<Integer>) s[3];
long t0 = System.nanoTime();
long[] slow = slowConvertDpp(fn, af, jk, dk);
long slowNs = System.nanoTime() - t0;
t0 = System.nanoTime();
long[] fast = fastConvertDpp(fn, af, jk, dk);
long fastNs = System.nanoTime() - t0;
double speedup = (double) slowNs / Math.max(fastNs, 1);
System.out.printf("T6: F=%d A=%d K=%d — slow=%.3fms fast=%.3fms speedup=%.1fx%n",
F, A, K, slowNs / 1e6, fastNs / 1e6, speedup);
long[] slow2 = slowConvertDpp(fn, af, jk, dk);
long[] fast2 = fastConvertDpp(fn, af, jk, dk);
test("T6: results match (F=1000,A=100,K=100)", slow[0] == fast[0]);
test("T6: slow ops >= A*F/2", slow2[1] >= (long) A * F / 2);
}
System.out.println();
System.out.println("6/6 PASS — flink-0005: DynamicPartitionPruningUtils " +
"indexOf+contains O(A*F + K*A) → HashMap+HashSet O(F + A + K)");
}
}