wave10a: 462/209 hudi/iceberg/beam-clean/samza-clean

defects/hudi/patch/hudi-0001-appendloadedinstants-list-contains.md

@@ -0,0 +1,67 @@
+# hudi-0001 — BaseHoodieTimeline.appendLoadedInstants List.contains O(N×M)
+
+## File
+`hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java`
+Line 123–131
+
+## Defect
+
+```java
+protected void appendLoadedInstants(List<HoodieInstant> loadedInstants) {
+    List<HoodieInstant> existingInstants = getInstants();          // List<HoodieInstant>
+    List<HoodieInstant> newInstants = loadedInstants.stream()
+        .filter(instant -> !existingInstants.contains(instant))   // O(M) per call
+        .collect(Collectors.toList());
+    if (!newInstants.isEmpty()) {
+      appendInstants(newInstants);
+    }
+}
+```
+
+`existingInstants` is a `List<HoodieInstant>` (the backing field `private List<HoodieInstant> instants`).
+`List.contains()` performs a linear scan — O(M) per invocation.
+The stream filter calls it once per element of `loadedInstants` — N calls total.
+
+Total complexity: **O(N × M)** where N = loadedInstants.size(), M = existing timeline size.
+
+In a Hudi table with a long history (thousands of commits), M can be large. This method is called
+during incremental timeline loading (time-range and limit-based), so both N and M can be in the
+thousands during compaction or restore operations.
+
+## Fix
+
+Convert `existingInstants` to a `HashSet` before the filter. `HoodieInstant` implements `equals()`
+and `hashCode()` based on timestamp+action+state — so a HashSet works correctly.
+
+```java
+protected void appendLoadedInstants(List<HoodieInstant> loadedInstants) {
+    Set<HoodieInstant> existingSet = new HashSet<>(getInstants());  // O(M) one-time
+    List<HoodieInstant> newInstants = loadedInstants.stream()
+        .filter(instant -> !existingSet.contains(instant))          // O(1) per call
+        .collect(Collectors.toList());
+    if (!newInstants.isEmpty()) {
+      appendInstants(newInstants);
+    }
+}
+```
+
+## Complexity
+
+| | Before | After |
+|-|--------|-------|
+| per-element membership test | O(M) | O(1) |
+| full filter pass | O(N × M) | O(N + M) |
+
+At N=M=10 000: 100 000 000 comparisons → 20 000 comparisons. **5000x fewer operations.**
+
+## Severity
+
+MEDIUM — triggered during incremental timeline load on large tables (restore, compaction, archival).
+Degrades linearly with table history depth.
+
+## Import required
+
+```java
+import java.util.HashSet;
+import java.util.Set;
+```

defects/hudi/patch/hudi-0002-pruneinternalschema-list-contains.md

@@ -0,0 +1,76 @@
+# hudi-0002 — InternalSchemaUtils.pruneInternalSchema ArrayList.contains O(N²) + pruneType O(F×D)
+
+## File
+`hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/InternalSchemaUtils.java`
+Lines 66–72 (topParentFieldIds dedup) and 105–160 (pruneType)
+
+## Defect — Part A: topParentFieldIds dedup (O(N²))
+
+```java
+List<Integer> topParentFieldIds = new ArrayList<>();
+names.stream().forEach(f -> {
+    int id = schema.findIdByName(f.split("\\.")[0]);
+    if (!topParentFieldIds.contains(id)) {      // O(N) scan of ArrayList per call
+        topParentFieldIds.add(id);
+    }
+});
+```
+
+`topParentFieldIds` is `ArrayList<Integer>`. `.contains(id)` is O(N) — called once per name.
+Total: O(N²) where N = names.size(). For a schema with many projected columns this becomes quadratic.
+
+## Defect — Part B: pruneType field membership (O(F×D))
+
+```java
+private static Type pruneType(Type type, List<Integer> fieldIds) {
+    // ...RECORD case:
+    for (Types.Field f : fields) {
+        Type newType = pruneType(f.type(), fieldIds);
+        if (fieldIds.contains(f.fieldId())) {    // O(D) scan per field
+            newTypes.add(f.type());
+        }
+    }
+    // ...ARRAY case:
+    if (fieldIds.contains(array.elementId())) {  // O(D) per array
+    // ...MAP case:
+    if (fieldIds.contains(map.valueId())) {      // O(D) per map
+```
+
+`fieldIds` is `List<Integer>`. Called recursively over the full schema tree (F nodes).
+Total: O(F × D) where F = total schema fields, D = projected field count.
+
+## Fix
+
+```java
+// Part A: use LinkedHashSet to preserve insertion order and deduplicate in O(1)
+Set<Integer> topParentFieldIdSet = new LinkedHashSet<>();
+names.stream().forEach(f -> {
+    int id = schema.findIdByName(f.split("\\.")[0]);
+    topParentFieldIdSet.add(id);   // HashSet.add deduplicates — O(1) amortized
+});
+List<Integer> topParentFieldIds = new ArrayList<>(topParentFieldIdSet);
+```
+
+```java
+// Part B: convert fieldIds to HashSet before entering pruneType
+private static Type pruneType(Type type, Set<Integer> fieldIds) {
+    // ...same logic, but fieldIds.contains() is O(1)
+}
+// Call site: pruneType(schema.getRecord(), new HashSet<>(fieldIds))
+```
+
+## Complexity
+
+| | Before | After |
+|-|--------|-------|
+| topParentFieldIds dedup | O(N²) | O(N) |
+| pruneType per-field check | O(D) | O(1) |
+| full pruneType traversal | O(F × D) | O(F) |
+
+At F=D=500 fields: 250 000 comparisons → 500. **500x fewer operations.**
+
+## Severity
+
+MEDIUM — triggered on every call to `pruneInternalSchema()`, which is called during query
+projection pushdown, Spark read, and schema evolution. Schemas with many nested columns amplify
+both defects simultaneously.

defects/hudi/patch/hudi-0003-metadatautil-logfilepaths-list-contains.md

@@ -0,0 +1,62 @@
+# hudi-0003 — HoodieTableMetadataUtil.getRevivedAndDeletedKeysFromMergedLogs List.contains O(N×M)
+
+## File
+`hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java`
+Line 1005–1007
+
+## Defect
+
+```java
+public static <T> Pair<Set<String>, Set<String>> getRevivedAndDeletedKeysFromMergedLogs(
+    ..., List<String> logFilePaths, ..., List<String> currentLogFilePaths, ...) {
+
+    List<String> logFilePathsWithoutCurrentLogFiles = logFilePaths.stream()
+        .filter(logFilePath -> !currentLogFilePaths.contains(logFilePath))  // O(M) per element
+        .collect(toList());
+```
+
+`currentLogFilePaths` is a `List<String>` (constructed at line 925 via `Collectors.toList()`).
+`List.contains()` does a linear scan — O(M) per call where M = currentLogFilePaths.size().
+The stream filter applies this once per element of `logFilePaths` — N calls total.
+
+Total complexity: **O(N × M)** where N = total log file paths, M = current log file paths count.
+
+This method is called during every Record-Level Index (RLI) update, which happens on each delta
+commit. In a table with many log files per partition (large MOR tables, frequent compaction), both
+N and M can be in the hundreds.
+
+## Fix
+
+Convert `currentLogFilePaths` to a `HashSet<String>` before the filter. String equality is
+well-defined and path strings are unique identifiers.
+
+```java
+Set<String> currentLogFilePathSet = new HashSet<>(currentLogFilePaths);  // O(M) once
+List<String> logFilePathsWithoutCurrentLogFiles = logFilePaths.stream()
+    .filter(logFilePath -> !currentLogFilePathSet.contains(logFilePath))   // O(1) per element
+    .collect(toList());
+```
+
+The call site (line 925–927) creates `currentLogFilePaths` as a List and passes it to this method.
+Alternatively, build it as a `HashSet` at the call site.
+
+## Complexity
+
+| | Before | After |
+|-|--------|-------|
+| per-element membership test | O(M) | O(1) |
+| full filter pass | O(N × M) | O(N + M) |
+
+At N=M=500 log files: 250 000 comparisons → 1000. **250x fewer operations.**
+
+## Severity
+
+MEDIUM — triggered on every delta commit when RLI is enabled. MOR (Merge-on-Read) tables with
+high write frequency and many log files per filegroup amplify this significantly.
+
+## Import required
+
+```java
+import java.util.HashSet;
+import java.util.Set;
+```