Skip to content

Commit 6eca5b8

Browse files
CaideyipiCaideyipi
authored
[Pipe] Add SHOW PIPE MEMORY for named memory blocks (#18575)
* [Pipe] Add SHOW PIPE MEMORY * [Pipe] Add hierarchical memory diagnostics to SHOW PIPE MEMORY * [Pipe] Fix memory diagnostic allocation behavior * [Pipe] Address SHOW PIPE MEMORY review feedback * [Pipe] Separate peak and max memory diagnostics * [Pipe] Share SHOW PIPE MEMORY across models and report pipe ownership * [Test] Calculate information schema table count dynamically
1 parent 0c0116a commit 6eca5b8

58 files changed

Lines changed: 2442 additions & 200 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBShowPipeMemoryIT.java‎

Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.db.it;
21+
22+
import org.apache.iotdb.isession.SessionConfig;
23+
import org.apache.iotdb.it.env.EnvFactory;
24+
import org.apache.iotdb.it.env.MultiEnvFactory;
25+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
26+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
27+
import org.apache.iotdb.itbase.category.ClusterIT;
28+
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
29+
import org.apache.iotdb.itbase.category.TableClusterIT;
30+
import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
31+
import org.apache.iotdb.itbase.env.BaseEnv;
32+
33+
import org.awaitility.Awaitility;
34+
import org.junit.AfterClass;
35+
import org.junit.BeforeClass;
36+
import org.junit.Test;
37+
import org.junit.experimental.categories.Category;
38+
import org.junit.runner.RunWith;
39+
40+
import java.sql.Connection;
41+
import java.sql.ResultSet;
42+
import java.sql.ResultSetMetaData;
43+
import java.sql.SQLException;
44+
import java.sql.Statement;
45+
import java.util.Arrays;
46+
import java.util.HashSet;
47+
import java.util.Set;
48+
import java.util.concurrent.TimeUnit;
49+
50+
import static org.junit.Assert.assertEquals;
51+
import static org.junit.Assert.assertNull;
52+
import static org.junit.Assert.assertThrows;
53+
import static org.junit.Assert.assertTrue;
54+
55+
@RunWith(IoTDBTestRunner.class)
56+
@Category({
57+
LocalStandaloneIT.class,
58+
ClusterIT.class,
59+
TableLocalStandaloneIT.class,
60+
TableClusterIT.class
61+
})
62+
public class IoTDBShowPipeMemoryIT {
63+
64+
private static final String PIPE_NAME = "memory_owner";
65+
private static final String USER = "pipe_memory_user";
66+
private static final String PASSWORD = "password123456";
67+
private static final String[] COLUMN_NAMES = {
68+
"block_id",
69+
"name",
70+
"category",
71+
"pipe_name",
72+
"pipe_model",
73+
"memory_usage_in_bytes",
74+
"peak_memory_size_in_bytes",
75+
"max_memory_size_in_bytes",
76+
"allocation_time",
77+
"assigner",
78+
"parent_block_id",
79+
"hierarchy_level",
80+
"accounted_memory_usage_in_bytes"
81+
};
82+
83+
@BeforeClass
84+
public static void setUp() throws Exception {
85+
EnvFactory.getEnv().getConfig().getCommonConfig().setEnforceStrongPassword(false);
86+
// Reserve enough pipe memory for both pipes and the shared caches in the small test JVM.
87+
EnvFactory.getEnv().getConfig().getDataNodeConfig().setDataNodeMemoryProportion("3:3:1:1:9:0");
88+
EnvFactory.getEnv().initClusterEnvironment();
89+
try (Connection connection = EnvFactory.getEnv().getConnection();
90+
Statement statement = connection.createStatement()) {
91+
statement.execute("CREATE USER " + USER + " '" + PASSWORD + "'");
92+
}
93+
}
94+
95+
@AfterClass
96+
public static void tearDown() throws Exception {
97+
EnvFactory.getEnv().cleanClusterEnvironment();
98+
}
99+
100+
@Test
101+
public void testColumnsAndPrivilegesInBothModels() throws Exception {
102+
for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) {
103+
for (String dialect : Arrays.asList(BaseEnv.TREE_SQL_DIALECT, BaseEnv.TABLE_SQL_DIALECT)) {
104+
try (Connection connection =
105+
EnvFactory.getEnv()
106+
.getConnection(
107+
dataNode,
108+
SessionConfig.DEFAULT_USER,
109+
SessionConfig.DEFAULT_PASSWORD,
110+
dialect);
111+
Statement statement = connection.createStatement()) {
112+
assertPipeMemoryColumnsAndReadOwners(statement, "SHOW PIPE MEMORY");
113+
}
114+
try (Connection connection =
115+
EnvFactory.getEnv().getConnection(dataNode, USER, PASSWORD, dialect);
116+
Statement statement = connection.createStatement()) {
117+
assertThrows(SQLException.class, () -> statement.executeQuery("SHOW PIPE MEMORY"));
118+
}
119+
}
120+
}
121+
try (Connection connection = EnvFactory.getEnv().getConnection();
122+
Statement statement = connection.createStatement()) {
123+
statement.execute("GRANT SYSTEM ON root.** TO USER " + USER);
124+
try {
125+
for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) {
126+
for (String dialect :
127+
Arrays.asList(BaseEnv.TREE_SQL_DIALECT, BaseEnv.TABLE_SQL_DIALECT)) {
128+
try (Connection userConnection =
129+
EnvFactory.getEnv().getConnection(dataNode, USER, PASSWORD, dialect);
130+
Statement userStatement = userConnection.createStatement()) {
131+
assertPipeMemoryColumnsAndReadOwners(userStatement, "SHOW PIPE MEMORY");
132+
}
133+
}
134+
}
135+
} finally {
136+
statement.execute("REVOKE SYSTEM ON root.** FROM USER " + USER);
137+
}
138+
}
139+
}
140+
141+
@Test
142+
public void testTreeAndTablePipeOwnership() throws Exception {
143+
MultiEnvFactory.createEnv(1);
144+
final BaseEnv receiver = MultiEnvFactory.getEnv(0);
145+
receiver.getConfig().getCommonConfig().setIsPipeEnableMemoryCheck(false);
146+
receiver.initClusterEnvironment(1, 1);
147+
try (Connection treeConnection = EnvFactory.getEnv().getConnection(BaseEnv.TREE_SQL_DIALECT);
148+
Statement treeStatement = treeConnection.createStatement();
149+
Connection tableConnection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
150+
Statement tableStatement = tableConnection.createStatement()) {
151+
treeStatement.execute("CREATE DATABASE root.pipe_memory");
152+
treeStatement.execute("CREATE TIMESERIES root.pipe_memory.d.s INT32");
153+
treeStatement.execute("INSERT INTO root.pipe_memory.d(time, s) VALUES (0, 0)");
154+
tableStatement.execute("CREATE DATABASE pipe_memory_db");
155+
tableStatement.execute("CREATE TABLE pipe_memory_db.t (device STRING TAG, s INT32 FIELD)");
156+
tableStatement.execute("INSERT INTO pipe_memory_db.t(time, device, s) VALUES (0, 'd', 0)");
157+
final String createPipe =
158+
"CREATE PIPE "
159+
+ PIPE_NAME
160+
+ " WITH SOURCE ('source.realtime.mode'='forced-log', 'source.history.enable'='false')"
161+
+ " WITH SINK ('sink'='iotdb-thrift-sync-sink', 'sink.ip'='"
162+
+ receiver.getDataNodeWrapper(0).getIp()
163+
+ "', 'sink.port'='"
164+
+ receiver.getDataNodeWrapper(0).getPort()
165+
+ "',"
166+
+ " 'sink.batch.enable'='false')";
167+
try {
168+
treeStatement.execute(createPipe);
169+
tableStatement.execute(createPipe);
170+
// Complete the handshake before stopping the receiver, so each pipe retains its events.
171+
receiver.shutdownDataNode(0);
172+
treeStatement.execute("INSERT INTO root.pipe_memory.d(time, s) VALUES (1, 1)");
173+
tableStatement.execute("INSERT INTO pipe_memory_db.t(time, device, s) VALUES (1, 'd', 1)");
174+
final Set<String> expectedOwners = new HashSet<>(Arrays.asList("tree", "table"));
175+
Awaitility.await()
176+
.atMost(60, TimeUnit.SECONDS)
177+
.untilAsserted(
178+
() -> {
179+
assertTrue(
180+
readPipeOwnersOnAllDataNodes(BaseEnv.TREE_SQL_DIALECT, "SHOW PIPE MEMORY")
181+
.containsAll(expectedOwners));
182+
assertTrue(
183+
readPipeOwnersOnAllDataNodes(BaseEnv.TABLE_SQL_DIALECT, "SHOW PIPE MEMORY")
184+
.containsAll(expectedOwners));
185+
assertTrue(
186+
readPipeOwnersOnAllDataNodes(
187+
BaseEnv.TABLE_SQL_DIALECT,
188+
"SELECT * FROM information_schema.pipe_memory")
189+
.containsAll(expectedOwners));
190+
});
191+
} finally {
192+
treeStatement.execute("DROP PIPE IF EXISTS " + PIPE_NAME);
193+
tableStatement.execute("DROP PIPE IF EXISTS " + PIPE_NAME);
194+
}
195+
} finally {
196+
receiver.cleanClusterEnvironment();
197+
}
198+
}
199+
200+
private static Set<String> readPipeOwnersOnAllDataNodes(String dialect, String sql)
201+
throws SQLException {
202+
final Set<String> owners = new HashSet<>();
203+
// Query each node separately because memory diagnostics describe local allocations.
204+
for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) {
205+
try (Connection connection =
206+
EnvFactory.getEnv()
207+
.getConnection(
208+
dataNode,
209+
SessionConfig.DEFAULT_USER,
210+
SessionConfig.DEFAULT_PASSWORD,
211+
dialect);
212+
Statement statement = connection.createStatement()) {
213+
owners.addAll(assertPipeMemoryColumnsAndReadOwners(statement, sql));
214+
}
215+
}
216+
return owners;
217+
}
218+
219+
private static Set<String> assertPipeMemoryColumnsAndReadOwners(Statement statement, String sql)
220+
throws SQLException {
221+
final Set<String> owners = new HashSet<>();
222+
try (ResultSet resultSet = statement.executeQuery(sql)) {
223+
final ResultSetMetaData metadata = resultSet.getMetaData();
224+
assertEquals(COLUMN_NAMES.length, metadata.getColumnCount());
225+
for (int i = 0; i < COLUMN_NAMES.length; i++) {
226+
assertEquals(COLUMN_NAMES[i], metadata.getColumnName(i + 1));
227+
}
228+
boolean hasFloatingMemory = false;
229+
while (resultSet.next()) {
230+
if ("FloatingMemory".equals(resultSet.getString("name"))) {
231+
assertNull(resultSet.getObject("pipe_name"));
232+
assertNull(resultSet.getObject("pipe_model"));
233+
assertTrue(resultSet.getLong("memory_usage_in_bytes") >= 0);
234+
hasFloatingMemory = true;
235+
}
236+
if (PIPE_NAME.equals(resultSet.getString("pipe_name"))) {
237+
final String model = resultSet.getString("pipe_model");
238+
assertTrue("tree".equals(model) || "table".equals(model));
239+
owners.add(model);
240+
}
241+
}
242+
assertTrue(hasFloatingMemory);
243+
}
244+
return owners;
245+
}
246+
}

‎integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java‎

Lines changed: 66 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -536,6 +536,7 @@ public void testInformationSchema() throws SQLException {
536536
"functions,INF,",
537537
"keywords,INF,",
538538
"nodes,INF,",
539+
"pipe_memory,INF,",
539540
"pipe_plugins,INF,",
540541
"pipes,INF,",
541542
"queries,INF,",
@@ -618,6 +619,24 @@ public void testInformationSchema() throws SQLException {
618619
"estimated_remaining_seconds,DOUBLE,ATTRIBUTE,",
619620
"is_degraded,BOOLEAN,ATTRIBUTE,",
620621
"recent_failures,STRING,ATTRIBUTE,")));
622+
TestUtils.assertResultSetEqual(
623+
statement.executeQuery("desc pipe_memory"),
624+
"ColumnName,DataType,Category,",
625+
new HashSet<>(
626+
Arrays.asList(
627+
"block_id,INT64,TAG,",
628+
"name,STRING,TAG,",
629+
"category,STRING,TAG,",
630+
"pipe_name,STRING,ATTRIBUTE,",
631+
"pipe_model,STRING,ATTRIBUTE,",
632+
"memory_usage_in_bytes,INT64,ATTRIBUTE,",
633+
"peak_memory_size_in_bytes,INT64,ATTRIBUTE,",
634+
"max_memory_size_in_bytes,INT64,ATTRIBUTE,",
635+
"allocation_time,TIMESTAMP,ATTRIBUTE,",
636+
"assigner,STRING,ATTRIBUTE,",
637+
"parent_block_id,INT64,ATTRIBUTE,",
638+
"hierarchy_level,INT32,ATTRIBUTE,",
639+
"accounted_memory_usage_in_bytes,INT64,ATTRIBUTE,")));
621640
TestUtils.assertResultSetEqual(
622641
statement.executeQuery("desc receivers"),
623642
"ColumnName,DataType,Category,",
@@ -751,6 +770,9 @@ public void testInformationSchema() throws SQLException {
751770
Assert.assertThrows(SQLException.class, () -> statement.execute("select * from data_nodes"));
752771
Assert.assertThrows(
753772
SQLException.class, () -> statement.executeQuery("select * from pipe_plugins"));
773+
Assert.assertThrows(
774+
SQLException.class, () -> statement.executeQuery("select * from pipe_memory"));
775+
Assert.assertThrows(SQLException.class, () -> statement.executeQuery("SHOW PIPE MEMORY"));
754776
Assert.assertThrows(
755777
SQLException.class, () -> statement.executeQuery("select * from table_disk_usage"));
756778

@@ -785,6 +807,47 @@ public void testInformationSchema() throws SQLException {
785807
// Test table query
786808
statement.execute("use information_schema");
787809

810+
try (final ResultSet resultSet = statement.executeQuery("SHOW PIPE MEMORY")) {
811+
final ResultSetMetaData metaData = resultSet.getMetaData();
812+
assertEquals(13, metaData.getColumnCount());
813+
assertEquals("block_id", metaData.getColumnName(1));
814+
assertEquals("name", metaData.getColumnName(2));
815+
assertEquals("category", metaData.getColumnName(3));
816+
assertEquals("pipe_name", metaData.getColumnName(4));
817+
assertEquals("pipe_model", metaData.getColumnName(5));
818+
assertEquals("memory_usage_in_bytes", metaData.getColumnName(6));
819+
assertEquals("peak_memory_size_in_bytes", metaData.getColumnName(7));
820+
assertEquals("max_memory_size_in_bytes", metaData.getColumnName(8));
821+
assertEquals("allocation_time", metaData.getColumnName(9));
822+
assertEquals("assigner", metaData.getColumnName(10));
823+
assertEquals("parent_block_id", metaData.getColumnName(11));
824+
assertEquals("hierarchy_level", metaData.getColumnName(12));
825+
assertEquals("accounted_memory_usage_in_bytes", metaData.getColumnName(13));
826+
boolean hasFloatingMemory = false;
827+
while (resultSet.next()) {
828+
if ("FloatingMemory".equals(resultSet.getString(2))) {
829+
assertTrue(resultSet.getLong("memory_usage_in_bytes") >= 0);
830+
assertNull(resultSet.getObject("pipe_name"));
831+
assertNull(resultSet.getObject("pipe_model"));
832+
hasFloatingMemory = true;
833+
}
834+
}
835+
assertTrue(hasFloatingMemory);
836+
}
837+
try (final ResultSet resultSet =
838+
statement.executeQuery("select * from information_schema.pipe_memory")) {
839+
boolean hasFloatingMemory = false;
840+
while (resultSet.next()) {
841+
if ("FloatingMemory".equals(resultSet.getString(2))) {
842+
assertTrue(resultSet.getLong("memory_usage_in_bytes") >= 0);
843+
assertNull(resultSet.getObject("pipe_name"));
844+
assertNull(resultSet.getObject("pipe_model"));
845+
hasFloatingMemory = true;
846+
}
847+
}
848+
assertTrue(hasFloatingMemory);
849+
}
850+
788851
statement.execute("create database test");
789852
statement.execute(
790853
"create table test.test (a tag, b attribute, c int32 comment 'turbine') comment 'test'");
@@ -834,6 +897,7 @@ public void testInformationSchema() throws SQLException {
834897
"information_schema,columns,INF,USING,null,SYSTEM VIEW,false,",
835898
"information_schema,queries,INF,USING,null,SYSTEM VIEW,false,",
836899
"information_schema,regions,INF,USING,null,SYSTEM VIEW,false,",
900+
"information_schema,pipe_memory,INF,USING,null,SYSTEM VIEW,false,",
837901
"information_schema,topics,INF,USING,null,SYSTEM VIEW,false,",
838902
"information_schema,pipe_plugins,INF,USING,null,SYSTEM VIEW,false,",
839903
"information_schema,pipes,INF,USING,null,SYSTEM VIEW,false,",
@@ -853,10 +917,11 @@ public void testInformationSchema() throws SQLException {
853917
"information_schema,queries_costs_histogram,INF,USING,null,SYSTEM VIEW,false,",
854918
"test,test,INF,USING,test,BASE TABLE,true,",
855919
"test,view_table,100,USING,null,VIEW FROM TREE,true,")));
920+
// Include all system tables and the table and view created in the test database.
856921
TestUtils.assertResultSetEqual(
857922
statement.executeQuery("count devices from tables where status = 'USING'"),
858923
"count(devices),",
859-
Collections.singleton("25,"));
924+
Collections.singleton((InformationSchema.getSchemaTables().size() + 2) + ","));
860925
TestUtils.assertResultSetEqual(
861926
statement.executeQuery(
862927
"select * from columns where table_name = 'queries' or database = 'test'"),

‎iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,7 @@ keyWords
148148
| LOAD
149149
| LOCAL
150150
| LOCK
151+
| MEMORY
151152
| MERGE
152153
| METADATA
153154
| MIGRATE

‎iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ ddlStatement
5555
// ExternalService
5656
| createService | startService | stopService | dropService | showService
5757
// Pipe Task
58-
| createPipe | alterPipe | dropPipe | startPipe | stopPipe | showPipes | showReceivers
58+
| createPipe | alterPipe | dropPipe | startPipe | stopPipe | showPipeMemory | showPipes | showReceivers
5959
// Pipe Plugin
6060
| createPipePlugin | dropPipePlugin | showPipePlugins
6161
// Subscription
@@ -702,6 +702,10 @@ showPipes
702702
: SHOW ((PIPE pipeName=identifier) | PIPES (WHERE (CONNECTOR | SINK) USED BY pipeName=identifier)?)
703703
;
704704

705+
showPipeMemory
706+
: SHOW PIPE MEMORY
707+
;
708+
705709
showReceivers
706710
: SHOW RECEIVERS
707711
;

‎iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -523,6 +523,10 @@ LOCK
523523
: L O C K
524524
;
525525

526+
MEMORY
527+
: M E M O R Y
528+
;
529+
526530
MERGE
527531
: M E R G E
528532
;

0 commit comments

Comments
 (0)