Skip to content

Commit b7beebe

Browse files
Copilotwt0530
Copilot
and
wt0530
committed
Integrate PR #1825 memory management infrastructure with spill-to-disk implementations
Co-authored-by: wt0530 <30097211+wt0530@users.noreply.github.com>
1 parent 759984b commit b7beebe

40 files changed

Lines changed: 2549 additions & 16 deletions

‎dingo-common/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ dependencies {
2828
implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: 'jackson'.v()
2929
implementation group: 'com.fasterxml.jackson.module', name: 'jackson-module-afterburner', version: 'jackson'.v()
3030
implementation group: 'org.apache.commons', name: 'commons-pool2', version: 'commons-pool2'.v()
31+
implementation group: 'org.openjdk.jol', name: 'jol-core', version: '0.2'
3132
api group: 'io.dingodb', name: 'calcite-core', version: 'calcite'.v()
3233
implementation group: 'commons-lang', name: 'commons-lang', version: 'commons-lang'.v()
3334
api group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: 'jackson'.v()

‎dingo-common/src/main/java/io/dingodb/common/ExecutionContext.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/*
2+
* Copyright 2021 DataCanvas
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.dingodb.common;
18+
19+
import io.dingodb.common.memory.MemoryPool;
20+
import io.dingodb.common.memory.QueryMemoryPoolHolder;
21+
import lombok.Data;
22+
import lombok.EqualsAndHashCode;
23+
import lombok.Setter;
24+
25+
import java.util.Properties;
26+
27+
@EqualsAndHashCode(callSuper = true)
28+
@Data
29+
public class ExecutionContext extends ExecuteVariables{
30+
public ExecutionContext() {
31+
32+
}
33+
34+
public ExecutionContext(Properties properties) {
35+
this.iterationLimit = getIterationLimit(properties);
36+
this.isJoinConcurrency = isJoinConcurrency(properties);
37+
this.concurrencyLevel = getConcurrencyLevel(properties);
38+
this.isInsertCheckInplace = isInsertCheckInplace(properties);
39+
this.isExecutorShuffle = isExecutorShuffle(properties);
40+
this.queryId = properties.getProperty("queryId", null);
41+
}
42+
43+
private long maxTimeout;
44+
private boolean isSelect;
45+
@Setter
46+
private String queryId;
47+
@Setter
48+
private String user;
49+
@Setter
50+
private String host;
51+
52+
private String traceId;
53+
54+
private boolean innerSql;
55+
56+
private QueryMemoryPoolHolder memoryPoolHolder = new QueryMemoryPoolHolder();
57+
58+
public MemoryPool getMemoryPool() {
59+
return memoryPoolHolder.getQueryMemoryPool();
60+
}
61+
62+
public void setMemoryPool(MemoryPool memoryPool) {
63+
memoryPoolHolder.initQueryMemoryPool(memoryPool);
64+
}
65+
66+
public void renewMemoryPoolHolder() {
67+
memoryPoolHolder.destroy();
68+
this.memoryPoolHolder = new QueryMemoryPoolHolder();
69+
}
70+
71+
public void clearAllMemoryPool() {
72+
memoryPoolHolder.destroy();
73+
}
74+
}

‎dingo-common/src/main/java/io/dingodb/common/error/MemoryNotEnoughException.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* Copyright 2021 DataCanvas
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.dingodb.common.error;
18+
19+
import java.io.Serial;
20+
21+
public class MemoryNotEnoughException extends RuntimeException {
22+
23+
@Serial
24+
private static final long serialVersionUID = 1135046738477559351L;
25+
}

‎dingo-common/src/main/java/io/dingodb/common/memory/ApMemoryPool.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
/*
2+
* Copyright 2021 DataCanvas
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.dingodb.common.memory;
18+
19+
public class ApMemoryPool extends MemoryPool{
20+
public ApMemoryPool(String name, long minLimit, long maxLimit, MemoryPool parent) {
21+
super(name, maxLimit, parent, MemoryType.GENERAL_AP);
22+
}
23+
}

‎dingo-common/src/main/java/io/dingodb/common/memory/BlockingMemoryPool.java‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,191 @@
1+
/*
2+
* Copyright 2021 DataCanvas
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.dingodb.common.memory;
18+
19+
import com.google.common.util.concurrent.ListenableFuture;
20+
import com.google.common.util.concurrent.SettableFuture;
21+
import io.dingodb.common.log.LogUtils;
22+
import io.dingodb.common.mysql.scope.ScopeVariables;
23+
import lombok.extern.slf4j.Slf4j;
24+
25+
import javax.annotation.concurrent.GuardedBy;
26+
27+
import static com.google.common.base.Preconditions.checkState;
28+
29+
@Slf4j
30+
public abstract class BlockingMemoryPool extends MemoryPool {
31+
@GuardedBy("this")
32+
private long tryMinRequestSize = 0;
33+
34+
@GuardedBy("this")
35+
private long minRequestSize = 0;
36+
37+
private long maxRequestSize = 0;
38+
39+
@GuardedBy("this")
40+
private SettableFuture<?> settableFuture;
41+
42+
@GuardedBy("this")
43+
private SettableFuture<?> trySettableFuture;
44+
45+
@GuardedBy("this")
46+
private boolean needMemoryRevoking = false;
47+
48+
protected long maxElasticMemory;
49+
50+
protected boolean blockFlag;
51+
52+
public BlockingMemoryPool(String name, long maxLimit, MemoryPool parent, MemoryType memoryType) {
53+
super(name, maxLimit, parent, memoryType);
54+
setMaxElasticMemory(ScopeVariables.getMemoryRevokingThreshold());
55+
}
56+
57+
public void setMaxElasticMemory(double maxElasticThreshold) {
58+
this.maxElasticMemory = (long) (this.maxLimit * maxElasticThreshold);
59+
}
60+
61+
@Override
62+
public void setMaxLimit(long memoryLimit) {
63+
this.maxLimit = memoryLimit;
64+
setMaxElasticMemory(ScopeVariables.getMemoryRevokingThreshold());
65+
}
66+
67+
@Override
68+
protected ListenableFuture<?> block(ListenableFuture<?> parent, long size) {
69+
if (parent != null && !parent.isDone()) {
70+
if (inheritParentFuture()) {
71+
return parent;
72+
} else {
73+
return NOT_BLOCKED;
74+
}
75+
} else {
76+
if (this.getMemoryType() == MemoryType.GLOBAL) {
77+
long maxRequestSizeTmp = reservedBytes + revocableBytes;
78+
if (maxRequestSizeTmp > maxRequestSize) {
79+
maxRequestSize = maxRequestSizeTmp;
80+
}
81+
}
82+
if (reservedBytes + revocableBytes > maxElasticMemory && ScopeVariables.enableSpill()) {
83+
log.info("The query use much more memory for the memory pool: " + name);
84+
if (minRequestSize <= 0 || minRequestSize > size) {
85+
minRequestSize = size;
86+
}
87+
if (revocableBytes >= minRequestSize) {
88+
if (settableFuture == null || settableFuture.isDone()) {
89+
settableFuture = SettableFuture.create();
90+
this.blockFlag = true;
91+
}
92+
checkState(!settableFuture.isDone(), "future is already completed");
93+
this.needMemoryRevoking = true;
94+
requestMemoryRevoke();
95+
return settableFuture;
96+
} else {
97+
return NOT_BLOCKED;
98+
}
99+
} else {
100+
return NOT_BLOCKED;
101+
}
102+
}
103+
}
104+
105+
@Override
106+
protected void tryBlock(MemoryAllocateFuture allocFuture, long size, boolean reserved) {
107+
if (size > maxElasticMemory) {
108+
outOfMemory(fullName, getMemoryUsage(), size, maxElasticMemory, reserved);
109+
}
110+
111+
if (tryMinRequestSize <= 0 || tryMinRequestSize > size) {
112+
tryMinRequestSize = size;
113+
}
114+
115+
if ((revocableBytes >= tryMinRequestSize || reserved) && ScopeVariables.enableSpill()) {
116+
log.info("The query use much more memory for the memory pool: " + name);
117+
if (trySettableFuture == null || trySettableFuture.isDone()) {
118+
trySettableFuture = SettableFuture.create();
119+
this.blockFlag = true;
120+
}
121+
allocFuture.setAllocateFuture(trySettableFuture);
122+
this.needMemoryRevoking = true;
123+
requestMemoryRevoke();
124+
} else {
125+
outOfMemory(fullName, getMemoryUsage(), size, maxElasticMemory, reserved);
126+
}
127+
}
128+
129+
@Override
130+
protected synchronized void notifyBlockedQuery() {
131+
if (settableFuture == null && trySettableFuture == null) {
132+
return;
133+
}
134+
135+
long availableBytes = maxElasticMemory - reservedBytes - revocableBytes;
136+
if (availableBytes >= minRequestSize && settableFuture != null && !settableFuture.isDone()) {
137+
minRequestSize = 0;
138+
LogUtils.info(log, "block settable set null");
139+
settableFuture.set(null);
140+
}
141+
if (availableBytes >= tryMinRequestSize && trySettableFuture != null && !trySettableFuture.isDone()) {
142+
tryMinRequestSize = 0;
143+
trySettableFuture.set(null);
144+
}
145+
}
146+
147+
@Override
148+
protected long getTryMaxLimit() {
149+
return maxElasticMemory;
150+
}
151+
152+
protected boolean inheritParentFuture() {
153+
return true;
154+
}
155+
156+
@Override
157+
public void destroy() {
158+
if (settableFuture != null) {
159+
settableFuture.set(null);
160+
}
161+
162+
if (trySettableFuture != null) {
163+
trySettableFuture.set(null);
164+
}
165+
super.destroy();
166+
}
167+
168+
public synchronized boolean isNeedMemoryRevoking() {
169+
return needMemoryRevoking;
170+
}
171+
172+
public synchronized void resetNeedMemoryRevoking() {
173+
this.needMemoryRevoking = false;
174+
}
175+
176+
public ListenableFuture<?> getBlockedFuture() {
177+
return settableFuture;
178+
}
179+
180+
public SettableFuture<?> getTrySettableFuture() {
181+
return trySettableFuture;
182+
}
183+
184+
public long getMinRequestMemory() {
185+
return tryMinRequestSize + minRequestSize;
186+
}
187+
188+
public boolean isBlockFlag() {
189+
return blockFlag;
190+
}
191+
}

‎dingo-common/src/main/java/io/dingodb/common/memory/GlobalMemoryPool.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* Copyright 2021 DataCanvas
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.dingodb.common.memory;
18+
19+
import io.dingodb.common.mysql.scope.ScopeVariables;
20+
21+
import java.util.List;
22+
import java.util.concurrent.CopyOnWriteArrayList;
23+
24+
import static java.util.Objects.requireNonNull;
25+
26+
public class GlobalMemoryPool extends BlockingMemoryPool {
27+
private final List<MemoryPoolListener> listeners = new CopyOnWriteArrayList<>();
28+
29+
public GlobalMemoryPool(String name, long limit) {
30+
super(name, limit, null, MemoryType.GLOBAL);
31+
this.maxElasticMemory = this.maxLimit;
32+
}
33+
34+
@Override
35+
public void requestMemoryRevoke() {
36+
this.onMemoryReserved();
37+
}
38+
39+
public void addListener(MemoryPoolListener listener) {
40+
listeners.add(requireNonNull(listener, "listener cannot be null"));
41+
}
42+
43+
public void removeListener(MemoryPoolListener listener) {
44+
listeners.remove(requireNonNull(listener, "listener cannot be null"));
45+
}
46+
47+
private void onMemoryReserved() {
48+
listeners
49+
.forEach(listener -> listener.onMemoryReserved(this, ScopeVariables.getMemoryRevokingTarget()));
50+
}
51+
}

0 commit comments

Comments
 (0)