326 lines
13 KiB
Java
326 lines
13 KiB
Java
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)");
|
||
}
|
||
}
|