238 lines
9.1 KiB
Java
238 lines
9.1 KiB
Java
package unit;
|
||
|
||
import java.util.*;
|
||
|
||
/**
|
||
* flink-0004: DynamicSinkUtils UPDATE column resolution O(C×U) vs O(C+U)
|
||
*
|
||
* Simulates getUpdatedColumns() and projectColumnsForUpdate() from:
|
||
* flink-table/flink-table-planner/.../connectors/DynamicSinkUtils.java
|
||
*
|
||
* Standalone — no JUnit, no Flink deps.
|
||
*/
|
||
public class DynamicSinkUtilsAlgorithm {
|
||
|
||
// -----------------------------------------------------------------------
|
||
// Slow: List.contains inside loop (as in production code)
|
||
// -----------------------------------------------------------------------
|
||
|
||
static class SlowResult {
|
||
final long ops;
|
||
final List<String> updatedColumns;
|
||
final List<Integer> columnIndexes;
|
||
|
||
SlowResult(long ops, List<String> updatedColumns, List<Integer> columnIndexes) {
|
||
this.ops = ops;
|
||
this.updatedColumns = updatedColumns;
|
||
this.columnIndexes = columnIndexes;
|
||
}
|
||
}
|
||
|
||
static SlowResult getUpdatedColumnsSlow(List<String> allColumns, List<String> updatedColumnNames) {
|
||
long ops = 0;
|
||
List<String> result = new ArrayList<>();
|
||
for (String column : allColumns) { // O(C) loop
|
||
for (String upd : updatedColumnNames) { // O(U) scan simulating List.contains
|
||
ops++;
|
||
if (upd.equals(column)) {
|
||
result.add(column);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
return new SlowResult(ops, result, Collections.emptyList());
|
||
}
|
||
|
||
static SlowResult projectColumnsForUpdateSlow(
|
||
List<Integer> updatedIndexes, List<String> allColumnNames, List<String> updatedColumnNames) {
|
||
long ops = 0;
|
||
List<Integer> result = new ArrayList<>();
|
||
for (int index : updatedIndexes) {
|
||
String colName = allColumnNames.get(index);
|
||
// contains scan: O(U)
|
||
boolean found = false;
|
||
int foundIdx = -1;
|
||
for (int i = 0; i < updatedColumnNames.size(); i++) {
|
||
ops++;
|
||
if (updatedColumnNames.get(i).equals(colName)) {
|
||
found = true;
|
||
// indexOf scan: another O(U) — simulate both as done in production
|
||
}
|
||
}
|
||
// second pass simulating indexOf (production does both contains + indexOf)
|
||
for (int i = 0; i < updatedColumnNames.size(); i++) {
|
||
ops++;
|
||
if (updatedColumnNames.get(i).equals(colName)) {
|
||
foundIdx = i;
|
||
break;
|
||
}
|
||
}
|
||
if (found) {
|
||
result.add(foundIdx);
|
||
}
|
||
}
|
||
return new SlowResult(ops, Collections.emptyList(), result);
|
||
}
|
||
|
||
// -----------------------------------------------------------------------
|
||
// Fast: HashSet/HashMap for O(1) lookup
|
||
// -----------------------------------------------------------------------
|
||
|
||
static class FastResult {
|
||
final long ops;
|
||
final List<String> updatedColumns;
|
||
final List<Integer> columnIndexes;
|
||
|
||
FastResult(long ops, List<String> updatedColumns, List<Integer> columnIndexes) {
|
||
this.ops = ops;
|
||
this.updatedColumns = updatedColumns;
|
||
this.columnIndexes = columnIndexes;
|
||
}
|
||
}
|
||
|
||
static FastResult getUpdatedColumnsFast(List<String> allColumns, List<String> updatedColumnNames) {
|
||
long ops = 0;
|
||
// Build set: O(U)
|
||
Set<String> updatedSet = new HashSet<>();
|
||
for (String name : updatedColumnNames) {
|
||
ops++;
|
||
updatedSet.add(name);
|
||
}
|
||
List<String> result = new ArrayList<>();
|
||
for (String column : allColumns) { // O(C) loop
|
||
ops++;
|
||
if (updatedSet.contains(column)) { // O(1)
|
||
result.add(column);
|
||
}
|
||
}
|
||
return new FastResult(ops, result, Collections.emptyList());
|
||
}
|
||
|
||
static FastResult projectColumnsForUpdateFast(
|
||
List<Integer> updatedIndexes, List<String> allColumnNames, List<String> updatedColumnNames) {
|
||
long ops = 0;
|
||
// Build index map: O(U)
|
||
Map<String, Integer> updatedMap = new HashMap<>();
|
||
for (int i = 0; i < updatedColumnNames.size(); i++) {
|
||
ops++;
|
||
updatedMap.put(updatedColumnNames.get(i), i);
|
||
}
|
||
List<Integer> result = new ArrayList<>();
|
||
for (int index : updatedIndexes) { // O(I)
|
||
ops++;
|
||
String colName = allColumnNames.get(index);
|
||
Integer i = updatedMap.get(colName); // O(1) replaces contains + indexOf
|
||
if (i != null) {
|
||
result.add(i);
|
||
}
|
||
}
|
||
return new FastResult(ops, Collections.emptyList(), result);
|
||
}
|
||
|
||
// -----------------------------------------------------------------------
|
||
// Test harness
|
||
// -----------------------------------------------------------------------
|
||
|
||
public static void main(String[] args) {
|
||
int passed = 0;
|
||
int total = 0;
|
||
|
||
// Test parameters
|
||
int C = 300; // total columns
|
||
int U = 30; // updated columns
|
||
int I = 150; // updated indexes (subset of C)
|
||
|
||
// Build column lists
|
||
List<String> allColumns = new ArrayList<>();
|
||
for (int i = 0; i < C; i++) {
|
||
allColumns.add("col_" + i);
|
||
}
|
||
|
||
// Updated column names: every 10th column
|
||
List<String> updatedColumnNames = new ArrayList<>();
|
||
for (int i = 0; i < U; i++) {
|
||
updatedColumnNames.add("col_" + (i * 10));
|
||
}
|
||
|
||
// Updated indexes: first I columns
|
||
List<Integer> updatedIndexes = new ArrayList<>();
|
||
for (int i = 0; i < I; i++) {
|
||
updatedIndexes.add(i);
|
||
}
|
||
|
||
// ---- Test 1: getUpdatedColumns correctness ----
|
||
total++;
|
||
SlowResult slowGetCols = getUpdatedColumnsSlow(allColumns, updatedColumnNames);
|
||
FastResult fastGetCols = getUpdatedColumnsFast(allColumns, updatedColumnNames);
|
||
|
||
boolean colsMatch = slowGetCols.updatedColumns.equals(fastGetCols.updatedColumns);
|
||
if (colsMatch) {
|
||
System.out.println("PASS test1: getUpdatedColumns produces same results (" +
|
||
slowGetCols.updatedColumns.size() + " columns found)");
|
||
passed++;
|
||
} else {
|
||
System.out.println("FAIL test1: getUpdatedColumns mismatch: slow=" +
|
||
slowGetCols.updatedColumns.size() + " fast=" + fastGetCols.updatedColumns.size());
|
||
}
|
||
|
||
// ---- Test 2: getUpdatedColumns ops ratio ----
|
||
total++;
|
||
long slowOps1 = slowGetCols.ops;
|
||
long fastOps1 = fastGetCols.ops;
|
||
double ratio1 = (double) slowOps1 / fastOps1;
|
||
if (ratio1 >= 10.0) {
|
||
System.out.printf("PASS test2: getUpdatedColumns slow=%d ops, fast=%d ops, ratio=%.1fx%n",
|
||
slowOps1, fastOps1, ratio1);
|
||
passed++;
|
||
} else {
|
||
System.out.printf("FAIL test2: getUpdatedColumns ratio=%.1fx (need >=10x) slow=%d fast=%d%n",
|
||
ratio1, slowOps1, fastOps1);
|
||
}
|
||
|
||
// ---- Test 3: projectColumnsForUpdate correctness ----
|
||
total++;
|
||
SlowResult slowProject = projectColumnsForUpdateSlow(updatedIndexes, allColumns, updatedColumnNames);
|
||
FastResult fastProject = projectColumnsForUpdateFast(updatedIndexes, allColumns, updatedColumnNames);
|
||
|
||
boolean projectMatch = slowProject.columnIndexes.equals(fastProject.columnIndexes);
|
||
if (projectMatch) {
|
||
System.out.println("PASS test3: projectColumnsForUpdate produces same results (" +
|
||
slowProject.columnIndexes.size() + " entries)");
|
||
passed++;
|
||
} else {
|
||
System.out.println("FAIL test3: projectColumnsForUpdate mismatch: slow=" +
|
||
slowProject.columnIndexes.size() + " fast=" + fastProject.columnIndexes.size());
|
||
}
|
||
|
||
// ---- Test 4: projectColumnsForUpdate ops ratio ----
|
||
total++;
|
||
long slowOps2 = slowProject.ops;
|
||
long fastOps2 = fastProject.ops;
|
||
double ratio2 = (double) slowOps2 / fastOps2;
|
||
if (ratio2 >= 10.0) {
|
||
System.out.printf("PASS test4: projectColumnsForUpdate slow=%d ops, fast=%d ops, ratio=%.1fx%n",
|
||
slowOps2, fastOps2, ratio2);
|
||
passed++;
|
||
} else {
|
||
System.out.printf("FAIL test4: projectColumnsForUpdate ratio=%.1fx (need >=10x) slow=%d fast=%d%n",
|
||
ratio2, slowOps2, fastOps2);
|
||
}
|
||
|
||
// ---- Test 5: empty updatedColumns edge case ----
|
||
total++;
|
||
SlowResult slowEmpty = getUpdatedColumnsSlow(allColumns, Collections.emptyList());
|
||
FastResult fastEmpty = getUpdatedColumnsFast(allColumns, Collections.emptyList());
|
||
if (slowEmpty.updatedColumns.isEmpty() && fastEmpty.updatedColumns.isEmpty()) {
|
||
System.out.println("PASS test5: empty updatedColumnNames → empty result");
|
||
passed++;
|
||
} else {
|
||
System.out.println("FAIL test5: empty updatedColumnNames edge case");
|
||
}
|
||
|
||
System.out.println("\n" + passed + "/" + total + " PASS");
|
||
|
||
if (passed != total) {
|
||
System.exit(1);
|
||
}
|
||
}
|
||
}
|