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

// 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.

// 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.

4.2 KiB Raw Blame History Unescape Escape