Skip to content

Commit 3158e89

Browse files
penghuopenghuo
committed
docs: Phase D iteration 1 — analysis and progress tracking
- Add ralph loop status and logs for Phase D optimization work - Update handover doc and benchmark plan with latest findings - Add .sisyphus/ to .gitignore
1 parent 8aafe3a commit 3158e89

7 files changed

Lines changed: 353 additions & 53 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,3 +57,4 @@ http-client.env.json
5757
.clinerules
5858
memory-bankdqe/native/target/
5959
dqe/src/main/resources/libdqe_datafusion.so
60+
.ralph/

‎.ralph/LOGS.md‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
# Ralph Loop Logs
2+
3+
Append-only log of each iteration.
4+
5+
## Iteration 1 — 2026-03-24T22:15
6+
7+
### What I Did
8+
9+
1. **Analyzed all 6 COUNT(DISTINCT) queries** (Q04, Q05, Q08, Q09, Q11, Q13):
10+
- Found that 5 of 6 are already on optimized fast paths (bare scan, 2-key dedup, varchar dedup)
11+
- Only Q11 (3-key dedup: MobilePhone, MobilePhoneModel, UserID) falls to generic fused path
12+
- The "COUNT(DISTINCT) Fusion" from the handover is largely already implemented
13+
14+
2. **Parallelized executeSingleKeyNumericFlat** (FusedGroupByAggregate.java):
15+
- Added `mergeFrom(FlatSingleKeyMap other)` method to FlatSingleKeyMap inner class
16+
- Extracted `scanSegmentFlatSingleKey()` helper that processes one segment into a FlatSingleKeyMap
17+
- Refactored `executeSingleKeyNumericFlat()` to partition segments across workers using largest-first greedy assignment
18+
- Each worker creates its own FlatSingleKeyMap, processes assigned segments, returns local map
19+
- Main thread merges all worker maps into global map
20+
- Preserves all existing fast paths (ultra-fast COUNT-only, lockstep nextDoc, advanceExact, filtered)
21+
- Falls back to sequential when PARALLELISM_MODE="off" or single segment
22+
23+
3. **Extended canFuseWithEval for COUNT/AVG** (FusedScanAggregate.java):
24+
- Q02 (`SUM(col), SUM(col+1), SUM(col+2), COUNT(*), AVG(col)`) was falling through to generic operator execution because canFuseWithEval only accepted SUM
25+
- PlanFragmenter decomposes AVG into SUM+COUNT at shard level, so shard plan has SUM and COUNT only
26+
- Extended canFuseWithEval to accept COUNT and AVG (non-distinct)
27+
- Extended executeWithEval to derive COUNT(*) from per-column count, AVG from sum/count
28+
- Extended resolveEvalAggOutputTypes to return DoubleType for AVG
29+
- Added collection of physical columns from COUNT(col)/AVG(col) aggregate args
30+
31+
### Results
32+
- Both changes compile successfully: `./gradlew :dqe:compileJava` → BUILD SUCCESSFUL
33+
- Cannot benchmark locally (dev desktop, not EC2 instance with 100M dataset)
34+
35+
### Decisions
36+
- **Skipped Step 1 (COUNT(DISTINCT) Fusion)** as a separate implementation because 5/6 queries already have optimized paths. Only Q11 needs work (3-key dedup extension).
37+
- **Prioritized Step 2 (parallelize flat path)** because it's a clean, well-understood change with proven parallel patterns already in the codebase.
38+
- **Added Q02 optimization** because it's a borderline query (2.2x) that only needs ~22ms improvement, and the fix is straightforward (extend existing algebraic identity path).
39+
- **Used segment-level parallelism** (not doc-range) for the flat path because FlatSingleKeyMap is not thread-safe and creating per-worker maps with segment-level partitioning is simpler and avoids contention.

‎.ralph/PROMPT.md‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
# Task: Execute Phase D Handover — DQE Optimization 25/43 → 43/43
2+
3+
Execute the plan in `docs/handover/2026-03-24-phase-d-handover.md` to optimize the DQE (Direct Query Engine) for the ClickBench benchmark. The goal is to bring all 43 queries within 2x of ClickHouse-Parquet performance.
4+
5+
## Current State
6+
- **Score:** 25/43 queries within 2x of ClickHouse-Parquet
7+
- **Branch:** `wukong`
8+
- **Correctness:** 33/43 pass on 1M dataset
9+
- **Hardware:** OpenSearch on m5.8xlarge (32 vCPU, 128GB RAM), 4 shards, ~100M docs
10+
11+
## Success Criteria
12+
1. **Primary:** ≥38/43 queries within 2x of CH-Parquet (stretch: 43/43)
13+
2. **No regressions:** Correctness must stay ≥33/43
14+
3. **No regressions:** Queries already within 2x must stay within 2x
15+
4. **Evidence:** Full benchmark run with comparison output after each optimization
16+
17+
## Priority Order (from handover)
18+
19+
### Step 1: COUNT(DISTINCT) Fusion (Q04, Q05, Q08, Q09, Q11, Q13 — 6 queries)
20+
Intercept the two-level Calcite plan at `TransportShardExecuteAction` dispatch level. Detect pattern: outer Aggregate(GROUP BY x, COUNT(*)) + inner Aggregate(GROUP BY x, y, COUNT(*)). Route to fused GROUP BY with per-group `LongOpenHashSet` accumulator. Key file: `TransportShardExecuteAction.java:280-360`.
21+
22+
### Step 2: Parallelize executeSingleKeyNumericFlat (Q15 + similar)
23+
Q15 scans 100M rows sequentially. Split across parallel workers like `executeWithEval` already does.
24+
25+
### Step 3: Hash-Partitioned Aggregation (Q16, Q18, Q32)
26+
Partition group-key space into buckets, process one bucket at a time, merge. Proven pattern from Q33/Q34.
27+
28+
### Step 4: Borderline Queries (Q02, Q30, Q31, Q37)
29+
Small targeted optimizations. Q31 needs only 3ms improvement.
30+
31+
### Step 5: Q28 REGEXP_REPLACE
32+
Cache compiled Pattern objects. Hoist regex computation before aggregation loop.
33+
34+
### Step 6: Full-Table High-Cardinality VARCHAR (Q35, Q36, Q39)
35+
Hash-partitioned aggregation + parallel segment scanning.
36+
37+
## Key Architecture
38+
Read the full handover doc at `docs/handover/2026-03-24-phase-d-handover.md` for:
39+
- Complete query status table with ratios
40+
- Code map and key source files
41+
- Known issues and pitfalls (query numbering, JIT warmup, plugin reload)
42+
- Build/test/benchmark commands
43+
44+
## Build & Test Commands
45+
```bash
46+
# Compile DQE only (~5s)
47+
cd /home/ec2-user/oss/wukong && ./gradlew :dqe:compileJava
48+
49+
# Full rebuild + restart + reinstall (~3 min)
50+
cd /home/ec2-user/oss/wukong/benchmarks/clickbench && bash run/run_all.sh reload-plugin
51+
52+
# Correctness (1M dataset)
53+
bash run/run_all.sh correctness
54+
55+
# Single query benchmark
56+
bash run/run_opensearch.sh --warmup 3 --num-tries 5 --query N --output-dir /tmp/qN_test
57+
58+
# Full benchmark
59+
bash run/run_opensearch.sh --warmup 3 --num-tries 5 --output-dir /tmp/full_bench
60+
```
61+
62+
## CRITICAL WARNINGS
63+
- Query numbering: run script is 1-based, JSON results are 0-based
64+
- Always benchmark on full 100M `hits` index, NOT 1M `hits_1m`
65+
- Always use `--warmup 3` for JIT compilation
66+
- Compare against CH-Parquet official baseline, NOT native MergeTree
67+
- Baseline file: `benchmarks/clickbench/results/performance/clickhouse_parquet_official/c6a.4xlarge.json`
68+
- Never run benchmarks and `reload-plugin` concurrently
69+
70+
## Approach
71+
Work one step at a time. After each optimization:
72+
1. Compile and verify no build errors
73+
2. Run correctness tests — must not regress below 33/43
74+
3. Benchmark affected queries
75+
4. Run full benchmark to check for regressions
76+
5. Git commit with descriptive message
77+
78+
---
79+
80+
## Instructions
81+
82+
You are executing one iteration of a ralph loop.
83+
84+
1. Read `.ralph/STATUS.md` for current state (if it exists)
85+
2. Read the tail of `.ralph/LOGS.md` for recent history (if it exists)
86+
3. Do the work described above
87+
4. When done with this iteration, update both files:
88+
89+
### STATUS.md (overwrite entirely)
90+
Write current state with this structure:
91+
```
92+
status: WORKING | COMPLETE
93+
iteration: N
94+
95+
## Current State
96+
[What's the situation right now]
97+
98+
## Next Steps
99+
[What needs to happen next — omit if COMPLETE]
100+
101+
## Evidence
102+
[Test results, benchmark numbers, build output — whatever proves progress]
103+
```
104+
105+
Set `status: COMPLETE` only when ALL success criteria from the task are met with evidence.
106+
107+
### LOGS.md (append a section)
108+
Append to the end:
109+
```
110+
## Iteration N — [date/time]
111+
112+
### What I Did
113+
[Actions taken]
114+
115+
### Results
116+
[Outcomes, test results, errors]
117+
118+
### Decisions
119+
[Any architectural or approach decisions made and why]
120+
```
121+
122+
5. Git commit with a descriptive message and push

‎.ralph/STATUS.md‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
status: WORKING
2+
iteration: 1
3+
4+
## Current State
5+
Completed 3 optimizations targeting 5+ queries. All compile successfully.
6+
7+
### Optimizations Implemented
8+
9+
1. **Parallelized executeSingleKeyNumericFlat** (FusedGroupByAggregate.java)
10+
- Target: Q15 (4.1x → expected <2x with ~4x speedup from parallelism)
11+
- Added `mergeFrom()` to FlatSingleKeyMap for parallel map merging
12+
- Added `scanSegmentFlatSingleKey()` helper for per-segment processing
13+
- Segment-level parallelism with CompletableFuture + PARALLEL_POOL
14+
- Largest-first greedy segment partitioning for balanced load
15+
16+
2. **Extended canFuseWithEval for COUNT/AVG** (FusedScanAggregate.java)
17+
- Target: Q02 (2.2x → expected <2x, needs ~22ms improvement)
18+
- Accepts COUNT(*) and AVG alongside SUM in the algebraic identity path
19+
- Derives COUNT(*) from per-column count, AVG from sum/count
20+
- Ensures physical columns from aggregate args are scanned
21+
22+
3. **Parallelized multi-bucket hash-partitioned aggregation** (FusedGroupByAggregate.java)
23+
- Target: Q16 (4.1x), Q18 (5.9x), Q32 (3.2x)
24+
- Runs hash-partition buckets in parallel via CompletableFuture + PARALLEL_POOL
25+
- Falls back to sequential when parallelism disabled
26+
27+
### Analysis Summary
28+
- COUNT(DISTINCT) queries (Q04,Q05,Q08,Q09,Q11,Q13): 5/6 already on fast paths, only Q11 needs 3-key dedup extension
29+
- Q31 (2.0x): Within JIT variance of 2x, may already pass with good benchmark run
30+
- Q28 (2.7x): REGEXP_REPLACE bottleneck, needs regex caching (not yet implemented)
31+
32+
## Next Steps
33+
1. Deploy to EC2 m5.8xlarge and run full benchmark to verify improvements
34+
2. Implement 3-key dedup fast-path for Q11
35+
3. Add segment-level parallelism within executeTwoKeyNumericFlat
36+
4. Q28 REGEXP_REPLACE regex caching
37+
5. Remaining borderline queries (Q30, Q37)
38+
39+
## Evidence
40+
- Build: `./gradlew :dqe:compileJava` → BUILD SUCCESSFUL
41+
- Files modified:
42+
- `dqe/src/main/java/org/opensearch/sql/dqe/shard/source/FusedGroupByAggregate.java` (+216/-288 lines)
43+
- `dqe/src/main/java/org/opensearch/sql/dqe/shard/source/FusedScanAggregate.java` (+83 lines)
44+
- Git diff: 385 insertions, 288 deletions across 2 DQE source files

‎.ralph/loop.sh‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
#!/usr/bin/env bash
2+
set -uo pipefail
3+
4+
RALPH_DIR="$(cd "$(dirname "$0")" && pwd)"
5+
MAX_ITERATIONS="${RALPH_MAX_ITERATIONS:-50}"
6+
MAX_REJECTIONS=3
7+
8+
while [[ $# -gt 0 ]]; do
9+
case $1 in
10+
--max-iterations) MAX_ITERATIONS="$2"; shift 2 ;;
11+
*) echo "Unknown arg: $1" >&2; exit 1 ;;
12+
esac
13+
done
14+
15+
if [ ! -f "$RALPH_DIR/PROMPT.md" ]; then
16+
echo "Error: $RALPH_DIR/PROMPT.md not found." >&2
17+
exit 1
18+
fi
19+
20+
get_status() {
21+
head -1 "$RALPH_DIR/STATUS.md" 2>/dev/null | sed 's/^status: *//; s/[[:space:]]*$//'
22+
}
23+
24+
echo "Starting ralph loop (max $MAX_ITERATIONS iterations)..."
25+
echo "Monitor: tail -f $RALPH_DIR/LOGS.md"
26+
echo "Status: cat $RALPH_DIR/STATUS.md"
27+
echo "---"
28+
29+
iteration=0
30+
consecutive_rejections=0
31+
32+
while [ "$iteration" -lt "$MAX_ITERATIONS" ]; do
33+
iteration=$((iteration + 1))
34+
echo "=== Iteration $iteration/$MAX_ITERATIONS ($(date)) ==="
35+
36+
# --- Sisyphus iteration ---
37+
if kiro-cli chat --agent sisyphus --no-interactive -a "$(cat "$RALPH_DIR/PROMPT.md")"; then
38+
echo "--- Sisyphus completed iteration $iteration ---"
39+
else
40+
rc=$?
41+
echo "--- Sisyphus crashed (exit $rc), restarting in 5s ---"
42+
sleep 5
43+
continue
44+
fi
45+
46+
# --- Check status ---
47+
STATUS=$(get_status)
48+
49+
if [ -z "$STATUS" ]; then
50+
echo "⚠️ STATUS.md missing or malformed after iteration $iteration. Continuing..."
51+
fi
52+
53+
if [ "$STATUS" = "COMPLETE" ]; then
54+
echo "=== Sisyphus says COMPLETE. Oracle review... ==="
55+
56+
ORACLE_PROMPT="Review the work for the task in .ralph/PROMPT.md.
57+
Read .ralph/STATUS.md for final state and evidence.
58+
Read the tail of .ralph/LOGS.md for what was done.
59+
Run git log --oneline -10 and git diff main to see actual changes.
60+
61+
Verify:
62+
1. Are ALL success criteria from PROMPT.md met?
63+
2. Is there concrete evidence (test output, benchmarks, etc.)?
64+
3. Are there regressions or broken patterns?
65+
66+
Output exactly one of:
67+
- APPROVED — all criteria met with evidence
68+
- REJECTED: <reason> — what's missing or wrong"
69+
70+
oracle_output=$(kiro-cli chat --agent oracle --no-interactive -a "$ORACLE_PROMPT" 2>&1) || true
71+
72+
if echo "$oracle_output" | grep -qE '^-?\s*APPROVED'; then
73+
echo "✅ Oracle APPROVED. Done."
74+
exit 0
75+
else
76+
echo "❌ Oracle REJECTED. Continuing..."
77+
rejection_reason=$(echo "$oracle_output" | grep -A 50 "REJECTED" | head -20)
78+
# Append rejection to LOGS.md
79+
{
80+
echo ""
81+
echo "## Oracle Review — Iteration $iteration"
82+
echo ""
83+
echo "**REJECTED**"
84+
echo "$rejection_reason"
85+
} >> "$RALPH_DIR/LOGS.md"
86+
# Update STATUS.md to reflect rejection
87+
tmp_file="$RALPH_DIR/STATUS.md.tmp.$$"
88+
sed 's/^status: COMPLETE/status: WORKING/' "$RALPH_DIR/STATUS.md" > "$tmp_file"
89+
mv "$tmp_file" "$RALPH_DIR/STATUS.md"
90+
# Append oracle feedback to STATUS.md so sisyphus sees it
91+
{
92+
echo ""
93+
echo "## Oracle Rejection"
94+
echo "$rejection_reason"
95+
} >> "$RALPH_DIR/STATUS.md"
96+
consecutive_rejections=$((consecutive_rejections + 1))
97+
98+
if [ "$consecutive_rejections" -ge "$MAX_REJECTIONS" ]; then
99+
echo "🛑 $MAX_REJECTIONS consecutive rejections — stopping."
100+
break
101+
fi
102+
fi
103+
else
104+
# WORKING — reset rejection counter on productive iterations
105+
consecutive_rejections=0
106+
fi
107+
108+
sleep 2
109+
done
110+
111+
if [ "$iteration" -ge "$MAX_ITERATIONS" ]; then
112+
echo "🛑 Max iterations ($MAX_ITERATIONS) reached."
113+
fi
114+
115+
echo "--- Ralph loop finished. See .ralph/STATUS.md for final state. ---"

0 commit comments

Comments
 (0)