1111// Some portions generated by Codex
1212package org .eclipse .rdf4j .query .algebra .evaluation .optimizer ;
1313
14+ import java .time .Clock ;
15+ import java .time .Duration ;
1416import java .util .Map ;
17+ import java .util .Objects ;
1518import java .util .concurrent .ConcurrentHashMap ;
19+ import java .util .concurrent .atomic .AtomicLong ;
1620import java .util .concurrent .atomic .LongAdder ;
1721
1822/**
1923 * In-memory statistics store for join pattern fanout metrics.
2024 */
2125public class MemoryJoinStats implements JoinStatsProvider {
2226
27+ public static final Duration DEFAULT_INVALIDATION_WINDOW = Duration .ofMinutes (10 );
28+ public static final long DEFAULT_INVALIDATION_STATEMENT_THRESHOLD = 100_000L ;
29+ public static final double DEFAULT_BASELINE_DRIFT_RATIO = 0.5d ;
30+
31+ public static final class InvalidationSettings {
32+ private final Duration window ;
33+ private final long statementThreshold ;
34+ private final double baselineDriftRatio ;
35+
36+ private InvalidationSettings (Duration window , long statementThreshold , double baselineDriftRatio ) {
37+ this .window = window ;
38+ this .statementThreshold = statementThreshold ;
39+ this .baselineDriftRatio = baselineDriftRatio ;
40+ }
41+
42+ public static InvalidationSettings disabled () {
43+ return new InvalidationSettings (Duration .ZERO , 0 , 0.0d );
44+ }
45+
46+ public static InvalidationSettings of (Duration window , long statementThreshold ) {
47+ return of (window , statementThreshold , DEFAULT_BASELINE_DRIFT_RATIO );
48+ }
49+
50+ public static InvalidationSettings of (Duration window , long statementThreshold , double baselineDriftRatio ) {
51+ Objects .requireNonNull (window , "window" );
52+ if (window .isNegative () || window .isZero ()) {
53+ throw new IllegalArgumentException ("window must be positive" );
54+ }
55+ if (statementThreshold <= 0 ) {
56+ throw new IllegalArgumentException ("statementThreshold must be positive" );
57+ }
58+ if (baselineDriftRatio < 0.0d ) {
59+ throw new IllegalArgumentException ("baselineDriftRatio must be >= 0" );
60+ }
61+ return new InvalidationSettings (window , statementThreshold , baselineDriftRatio );
62+ }
63+
64+ public Duration getWindow () {
65+ return window ;
66+ }
67+
68+ public long getStatementThreshold () {
69+ return statementThreshold ;
70+ }
71+
72+ public double getBaselineDriftRatio () {
73+ return baselineDriftRatio ;
74+ }
75+
76+ private boolean enabled () {
77+ return statementThreshold > 0 && !window .isZero ();
78+ }
79+ }
80+
2381 private static final class Stats {
2482 private final LongAdder calls = new LongAdder ();
2583 private final LongAdder results = new LongAdder ();
2684 private final long priorCalls ;
2785 private final double priorResults ;
86+ private final double baselineCardinality ;
2887
29- private Stats (long priorCalls , double priorResults ) {
88+ private Stats (long priorCalls , double priorResults , double baselineCardinality ) {
3089 this .priorCalls = priorCalls ;
3190 this .priorResults = priorResults ;
91+ this .baselineCardinality = baselineCardinality ;
3292 }
3393
3494 private double averageResults () {
@@ -43,9 +103,34 @@ private double averageResults() {
43103 private long actualCalls () {
44104 return calls .sum ();
45105 }
106+
107+ private double baselineCardinality () {
108+ return baselineCardinality ;
109+ }
46110 }
47111
48112 private final Map <PatternKey , Stats > stats = new ConcurrentHashMap <>();
113+ private final InvalidationSettings invalidationSettings ;
114+ private final Clock clock ;
115+ private final Object invalidationLock = new Object ();
116+ private final AtomicLong windowStartMillis ;
117+ private final LongAdder statementsAddedInWindow = new LongAdder ();
118+
119+ public MemoryJoinStats () {
120+ this (InvalidationSettings .of (DEFAULT_INVALIDATION_WINDOW , DEFAULT_INVALIDATION_STATEMENT_THRESHOLD ),
121+ Clock .systemUTC ());
122+ }
123+
124+ public MemoryJoinStats (InvalidationSettings invalidationSettings ) {
125+ this (invalidationSettings , Clock .systemUTC ());
126+ }
127+
128+ public MemoryJoinStats (InvalidationSettings invalidationSettings , Clock clock ) {
129+ this .invalidationSettings = Objects .requireNonNull (invalidationSettings , "invalidationSettings" );
130+ this .clock = Objects .requireNonNull (clock , "clock" );
131+ long start = invalidationSettings .enabled () ? clock .millis () : 0L ;
132+ this .windowStartMillis = new AtomicLong (start );
133+ }
49134
50135 @ Override
51136 public void reset () {
@@ -54,22 +139,30 @@ public void reset() {
54139
55140 @ Override
56141 public void recordCall (PatternKey key ) {
57- stats .computeIfAbsent (key , ignored -> new Stats (0 , 0 )).calls .increment ();
142+ stats .computeIfAbsent (key , ignored -> new Stats (0 , 0 , 0.0d )).calls .increment ();
58143 }
59144
60145 @ Override
61146 public void recordResults (PatternKey key , long resultCount ) {
62147 if (resultCount < 0 ) {
63148 resultCount = 0 ;
64149 }
65- stats .computeIfAbsent (key , ignored -> new Stats (0 , 0 )).results .add (resultCount );
150+ stats .computeIfAbsent (key , ignored -> new Stats (0 , 0 , 0.0d )).results .add (resultCount );
66151 }
67152
68153 @ Override
69154 public void seedIfAbsent (PatternKey key , double defaultCardinality , long priorCalls ) {
70155 long seedCalls = Math .max (0 , priorCalls );
71156 double priorResults = Math .max (0.0d , defaultCardinality ) * seedCalls ;
72- stats .computeIfAbsent (key , ignored -> new Stats (seedCalls , priorResults ));
157+ stats .compute (key , (ignored , existing ) -> {
158+ if (existing == null ) {
159+ return new Stats (seedCalls , priorResults , defaultCardinality );
160+ }
161+ if (baselineDrifted (existing , defaultCardinality )) {
162+ return new Stats (seedCalls , priorResults , defaultCardinality );
163+ }
164+ return existing ;
165+ });
73166 }
74167
75168 @ Override
@@ -91,4 +184,45 @@ public long getTotalCalls() {
91184 }
92185 return total ;
93186 }
187+
188+ @ Override
189+ public void recordStatementsAdded (long statementCount ) {
190+ if (!invalidationSettings .enabled () || statementCount <= 0 ) {
191+ return ;
192+ }
193+ synchronized (invalidationLock ) {
194+ long nowMillis = clock .millis ();
195+ long windowStart = windowStartMillis .get ();
196+ long windowMillis = invalidationSettings .getWindow ().toMillis ();
197+ if (windowMillis <= 0 ) {
198+ return ;
199+ }
200+ if (nowMillis - windowStart >= windowMillis ) {
201+ windowStartMillis .set (nowMillis );
202+ statementsAddedInWindow .reset ();
203+ }
204+ statementsAddedInWindow .add (statementCount );
205+ if (statementsAddedInWindow .sum () >= invalidationSettings .getStatementThreshold ()) {
206+ reset ();
207+ windowStartMillis .set (nowMillis );
208+ statementsAddedInWindow .reset ();
209+ }
210+ }
211+ }
212+
213+ private boolean baselineDrifted (Stats existing , double newBaseline ) {
214+ double driftRatio = invalidationSettings .getBaselineDriftRatio ();
215+ if (driftRatio <= 0.0d ) {
216+ return false ;
217+ }
218+ double baseline = existing .baselineCardinality ();
219+ if (baseline <= 0.0d ) {
220+ return false ;
221+ }
222+ if (newBaseline <= 0.0d ) {
223+ return baseline > 0.0d ;
224+ }
225+ double relativeChange = Math .abs (newBaseline - baseline ) / baseline ;
226+ return relativeChange >= driftRatio ;
227+ }
94228}
0 commit comments