# UNDF: UNDF-2026-000000398
# 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.
