Skip to content

Commit fc2e0c1

Browse files
cambyzjucambyzju
authored
[BugFix] Be crash while query information_schema.warehouse_queries (backport #72019) (#72394)
1 parent d0a372f commit fc2e0c1

11 files changed

Lines changed: 132 additions & 24 deletions

File tree

‎be/src/exec/schema_scanner/schema_warehouse_metrics.cpp‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ namespace starrocks {
2626

2727
SchemaScanner::ColumnDesc WarehouseMetricsScanner::_s_columns[] = {
2828
// name, type, size, is_null
29-
{"WAREHOUSE_ID", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
29+
{"WAREHOUSE_ID", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(int64_t), false},
3030
{"WAREHOUSE_NAME", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
3131
{"QUEUE_PENDING_LENGTH", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
3232
{"QUEUE_RUNNING_LENGTH", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
@@ -89,7 +89,15 @@ Status WarehouseMetricsScanner::get_next(ChunkPtr* chunk, bool* eos) {
8989
Status WarehouseMetricsScanner::fill_chunk(ChunkPtr* chunk) {
9090
auto& slot_id_map = (*chunk)->get_slot_id_to_index_map();
9191
const TGetWarehouseMetricsResponeItem& item = _response.metrics[_idx++];
92-
DatumArray datum_array{Slice(item.warehouse_id),
92+
// Parse WAREHOUSE_ID from thrift string to int64.
93+
int64_t warehouse_id = 0;
94+
try {
95+
warehouse_id = std::stoll(item.warehouse_id);
96+
} catch (const std::exception& e) {
97+
return Status::InternalError(
98+
fmt::format("invalid warehouse_id '{}' in warehouse_metrics: {}", item.warehouse_id, e.what()));
99+
}
100+
DatumArray datum_array{warehouse_id,
93101
Slice(item.warehouse_name),
94102
Slice(item.queue_pending_length),
95103
Slice(item.queue_running_length),
@@ -107,4 +115,4 @@ Status WarehouseMetricsScanner::fill_chunk(ChunkPtr* chunk) {
107115
}
108116
return {};
109117
}
110-
} // namespace starrocks
118+
} // namespace starrocks

‎be/src/exec/schema_scanner/schema_warehouse_queries.cpp‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ namespace starrocks {
2626

2727
SchemaScanner::ColumnDesc WarehouseQueriesScanner::_s_columns[] = {
2828
// name, type, size, is_null
29-
{"WAREHOUSE_ID", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
29+
{"WAREHOUSE_ID", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(int64_t), false},
3030
{"WAREHOUSE_NAME", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
3131
{"QUERY_ID", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
3232
{"STATE", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), false},
@@ -87,7 +87,15 @@ Status WarehouseQueriesScanner::get_next(ChunkPtr* chunk, bool* eos) {
8787
Status WarehouseQueriesScanner::fill_chunk(ChunkPtr* chunk) {
8888
auto& slot_id_map = (*chunk)->get_slot_id_to_index_map();
8989
const TGetWarehouseQueriesResponseItem& item = _response.queries[_idx++];
90-
DatumArray datum_array{Slice(item.warehouse_id),
90+
// Parse WAREHOUSE_ID from thrift string to int64.
91+
int64_t warehouse_id = 0;
92+
try {
93+
warehouse_id = std::stoll(item.warehouse_id);
94+
} catch (const std::exception& e) {
95+
return Status::InternalError(
96+
fmt::format("invalid warehouse_id '{}' in warehouse_queries: {}", item.warehouse_id, e.what()));
97+
}
98+
DatumArray datum_array{warehouse_id,
9199
Slice(item.warehouse_name),
92100
Slice(item.query_id),
93101
Slice(item.state),
@@ -105,4 +113,4 @@ Status WarehouseQueriesScanner::fill_chunk(ChunkPtr* chunk) {
105113
}
106114
return {};
107115
}
108-
} // namespace starrocks
116+
} // namespace starrocks

‎fe/fe-core/src/main/java/com/starrocks/qe/scheduler/warehouse/WarehouseMetrics.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ public TGetWarehouseMetricsResponeItem toThrift() {
9999

100100
public List<ScalarOperator> toConstantOperators() {
101101
List<ScalarOperator> result = Lists.newArrayList();
102-
result.add(ConstantOperator.createVarchar(String.valueOf(warehouseId)));
102+
result.add(ConstantOperator.createBigint(warehouseId));
103103
result.add(ConstantOperator.createVarchar(warehouseName));
104104
result.add(ConstantOperator.createVarchar(String.valueOf(queuePendingLength)));
105105
result.add(ConstantOperator.createVarchar(String.valueOf(queueRunningLength)));

‎fe/fe-core/src/main/java/com/starrocks/qe/scheduler/warehouse/WarehouseQueryMetrics.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,13 +94,13 @@ public TGetWarehouseQueriesResponseItem toThrift() {
9494

9595
public List<ScalarOperator> toConstantOperators() {
9696
List<ScalarOperator> result = Lists.newArrayList();
97-
result.add(ConstantOperator.createVarchar(String.valueOf(warehouseId)));
97+
result.add(ConstantOperator.createBigint(warehouseId));
9898
result.add(ConstantOperator.createVarchar(warehouseName));
9999
result.add(ConstantOperator.createVarchar(DebugUtil.printId(queryId)));
100100
result.add(ConstantOperator.createVarchar(state.name()));
101101
result.add(ConstantOperator.createVarchar(String.valueOf(estCostsSlots)));
102102
result.add(ConstantOperator.createVarchar(String.valueOf(allocateSlots)));
103-
result.add(ConstantOperator.createDouble(queuedWaitSeconds));
103+
result.add(ConstantOperator.createVarchar(String.valueOf(queuedWaitSeconds)));
104104
result.add(ConstantOperator.createVarchar(query));
105105
if (extraMessage.isPresent()) {
106106
LogicalSlot.ExtraMessage extra = extraMessage.get();

‎fe/fe-core/src/test/java/com/starrocks/qe/scheduler/warehouse/WarehouseMetricsTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,14 @@
1717

1818
package com.starrocks.qe.scheduler.warehouse;
1919

20+
import com.starrocks.catalog.Column;
21+
import com.starrocks.catalog.system.information.WarehouseMetricsSystemTable;
22+
import com.starrocks.sql.optimizer.operator.scalar.ConstantOperator;
23+
import com.starrocks.sql.optimizer.operator.scalar.ScalarOperator;
2024
import com.starrocks.thrift.TGetWarehouseMetricsResponeItem;
2125
import org.junit.jupiter.api.Test;
2226

27+
import java.util.List;
2328
import java.util.Optional;
2429

2530
import static org.assertj.core.api.Assertions.assertThat;
@@ -59,4 +64,23 @@ public void testCreateWarehouseMetrics() {
5964
assertThat(thrift.getRemain_slots()).isEqualTo("1");
6065
assertThat(thrift.getMax_slots()).isEqualTo("1");
6166
}
67+
68+
@Test
69+
public void testToConstantOperatorsTypesMatchSchema() {
70+
WarehouseMetrics metrics = new WarehouseMetrics(42L, "wh", 1, 1, 1, 1, 1, 1, 1, 1, 1,
71+
Optional.empty());
72+
List<ScalarOperator> ops = metrics.toConstantOperators();
73+
List<Column> schema = new WarehouseMetricsSystemTable().getFullSchema();
74+
assertThat(ops).hasSize(schema.size());
75+
for (int i = 0; i < schema.size(); i++) {
76+
Column col = schema.get(i);
77+
ScalarOperator op = ops.get(i);
78+
assertThat(op).isInstanceOf(ConstantOperator.class);
79+
assertThat(op.getType().getPrimitiveType())
80+
.as("column %s at index %d", col.getName(), i)
81+
.isEqualTo(col.getType().getPrimitiveType());
82+
}
83+
assertThat(((ConstantOperator) ops.get(0)).getBigint()).isEqualTo(42L);
84+
assertThat(((ConstantOperator) ops.get(1)).getVarchar()).isEqualTo("wh");
85+
}
6286
}

‎fe/fe-core/src/test/java/com/starrocks/qe/scheduler/warehouse/WarehouseQueryMetricsTest.java‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,18 @@
1717

1818
package com.starrocks.qe.scheduler.warehouse;
1919

20+
import com.starrocks.catalog.Column;
21+
import com.starrocks.catalog.system.information.WarehouseQueriesSystemTable;
2022
import com.starrocks.qe.scheduler.slot.LogicalSlot;
23+
import com.starrocks.sql.optimizer.operator.scalar.ConstantOperator;
24+
import com.starrocks.sql.optimizer.operator.scalar.ScalarOperator;
2125
import com.starrocks.thrift.TGetWarehouseQueriesResponseItem;
2226
import org.junit.jupiter.api.Test;
2327

24-
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
28+
import java.util.List;
29+
import java.util.Optional;
30+
31+
import static org.assertj.core.api.Assertions.assertThat;
2532

2633
public class WarehouseQueryMetricsTest {
2734
@Test
@@ -53,4 +60,25 @@ public void testCreateWarehouseQueryMetrics() {
5360
assertThat(thrift.getQueued_wait_seconds()).isEqualTo("1.0");
5461
assertThat(thrift.getQuery()).isEqualTo("select 1");
5562
}
63+
64+
@Test
65+
public void testToConstantOperatorsTypesMatchSchema() {
66+
WarehouseQueryMetrics metrics = new WarehouseQueryMetrics(42L, "wh",
67+
null, LogicalSlot.State.ALLOCATED, 1, 1, 1.5,
68+
"select 1", Optional.empty());
69+
List<ScalarOperator> ops = metrics.toConstantOperators();
70+
List<Column> schema = new WarehouseQueriesSystemTable().getFullSchema();
71+
assertThat(ops).hasSize(schema.size());
72+
for (int i = 0; i < schema.size(); i++) {
73+
Column col = schema.get(i);
74+
ScalarOperator op = ops.get(i);
75+
assertThat(op).isInstanceOf(ConstantOperator.class);
76+
assertThat(op.getType().getPrimitiveType())
77+
.as("column %s at index %d", col.getName(), i)
78+
.isEqualTo(col.getType().getPrimitiveType());
79+
}
80+
// Spot-check key column values round-trip through the correct type
81+
assertThat(((ConstantOperator) ops.get(0)).getBigint()).isEqualTo(42L);
82+
assertThat(((ConstantOperator) ops.get(1)).getVarchar()).isEqualTo("wh");
83+
}
5684
}

‎fe/fe-core/src/test/java/com/starrocks/service/InformationSchemaDataSourceTest.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -620,30 +620,30 @@ public Map<Long, BaseSlotTracker> getWarehouseIdToSlotTracker() {
620620
// supported
621621
starRocksAssert.query("select count(1) from information_schema.warehouse_metrics")
622622
.explainContains(" constant exprs: \n" +
623-
" '0'");
623+
" 0");
624624
starRocksAssert.query("select * from information_schema.warehouse_metrics")
625625
.explainContains(" constant exprs: \n" +
626-
" '0'");
626+
" 0");
627627
starRocksAssert.query("select WAREHOUSE_NAME, REMAIN_SLOTS from information_schema.warehouse_metrics")
628628
.explainContains("constant exprs: \n" +
629629
" 'default_warehouse' | '0'");
630630
starRocksAssert.query("select count(1) from information_schema.warehouse_metrics where warehouse_id = '0'")
631631
.explainContains(" constant exprs: \n" +
632-
" '0'");
632+
" 0");
633633
starRocksAssert.query("select * from information_schema.warehouse_metrics where WAREHOUSE_ID = '0'")
634634
.explainContains(" constant exprs: \n" +
635-
" '0'");
635+
" 0");
636636
starRocksAssert.query("select WAREHOUSE_NAME, REMAIN_SLOTS from information_schema.warehouse_metrics " +
637637
"where WAREHOUSE_ID = 0")
638638
.explainContains("constant exprs: \n" +
639639
" 'default_warehouse' | '0'");
640640
starRocksAssert.query("select count(1) from information_schema.warehouse_metrics where WAREHOUSE_NAME = " +
641641
"'default_warehouse'")
642642
.explainContains(" constant exprs: \n" +
643-
" '0'");
643+
" 0");
644644
starRocksAssert.query("select * from information_schema.warehouse_metrics where WAREHOUSE_NAME= 'default_warehouse'")
645645
.explainContains(" constant exprs: \n" +
646-
" '0'");
646+
" 0");
647647
starRocksAssert.query("select WAREHOUSE_NAME, REMAIN_SLOTS from information_schema.warehouse_metrics " +
648648
"where WAREHOUSE_NAME = 'default_warehouse'")
649649
.explainContains("constant exprs: \n" +
@@ -681,30 +681,30 @@ public List<LogicalSlot> getSlots() {
681681
// supported
682682
starRocksAssert.query("select count(1) from information_schema.warehouse_queries")
683683
.explainContains(" constant exprs: \n" +
684-
" '0'");
684+
" 0");
685685
starRocksAssert.query("select * from information_schema.warehouse_queries")
686686
.explainContains(" constant exprs: \n" +
687-
" '0'");
687+
" 0");
688688
starRocksAssert.query("select WAREHOUSE_NAME, EST_COSTS_SLOTS from information_schema.warehouse_queries")
689689
.explainContains(" constant exprs: \n" +
690690
" 'default_warehouse' | '1'");
691691
starRocksAssert.query("select count(1) from information_schema.warehouse_queries where warehouse_id = '0'")
692692
.explainContains(" constant exprs: \n" +
693-
" '0'");
693+
" 0");
694694
starRocksAssert.query("select * from information_schema.warehouse_queries where WAREHOUSE_ID = '0'")
695695
.explainContains(" constant exprs: \n" +
696-
" '0'");
696+
" 0");
697697
starRocksAssert.query("select WAREHOUSE_NAME, EST_COSTS_SLOTS from information_schema.warehouse_queries " +
698698
"where WAREHOUSE_ID = 0")
699699
.explainContains(" constant exprs: \n" +
700700
" 'default_warehouse' | '1'");
701701
starRocksAssert.query("select count(1) from information_schema.warehouse_queries where WAREHOUSE_NAME = " +
702702
"'default_warehouse'")
703703
.explainContains(" constant exprs: \n" +
704-
" '0'");
704+
" 0");
705705
starRocksAssert.query("select * from information_schema.warehouse_queries where WAREHOUSE_NAME= 'default_warehouse'")
706706
.explainContains(" constant exprs: \n" +
707-
" '0'");
707+
" 0");
708708
starRocksAssert.query("select WAREHOUSE_NAME, EST_COSTS_SLOTS from information_schema.warehouse_queries " +
709709
"where WAREHOUSE_NAME = 'default_warehouse'")
710710
.explainContains(" constant exprs: \n" +

‎test/sql/test_information_schema/R/test_warehouse_metrics‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,19 @@ EXTRA_MESSAGE varchar(2048) YES false None
2323
SELECT * FROM information_schema.warehouse_metrics;
2424
-- result:
2525
-- !result
26+
SET enable_evaluate_schema_scan_rule=false;
27+
-- result:
28+
-- !result
29+
SELECT * FROM information_schema.warehouse_metrics;
30+
-- result:
31+
-- !result
32+
SELECT COUNT(*) FROM information_schema.warehouse_metrics;
33+
-- result:
34+
0
35+
-- !result
36+
SET enable_evaluate_schema_scan_rule=true;
37+
-- result:
38+
-- !result
2639
DROP DATABASE db_${uuid0};
2740
-- result:
2841
-- !result

‎test/sql/test_information_schema/R/test_warehouse_queries‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,19 @@ EXTRA_MESSAGE varchar(2048) YES false None
2323
SELECT * FROM information_schema.warehouse_queries;
2424
-- result:
2525
-- !result
26+
SET enable_evaluate_schema_scan_rule=false;
27+
-- result:
28+
-- !result
29+
SELECT * FROM information_schema.warehouse_queries;
30+
-- result:
31+
-- !result
32+
SELECT COUNT(*) FROM information_schema.warehouse_queries;
33+
-- result:
34+
0
35+
-- !result
36+
SET enable_evaluate_schema_scan_rule=true;
37+
-- result:
38+
-- !result
2639
DROP DATABASE db_${uuid0};
2740
-- result:
2841
-- !result

‎test/sql/test_information_schema/T/test_warehouse_metrics‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,5 +2,12 @@
22
CREATE DATABASE db_${uuid0};
33
USE db_${uuid0};
44
desc information_schema.warehouse_metrics;
5+
-- FE-eval path (default): ConstantOperator types must match schema (WAREHOUSE_ID=BIGINT)
56
SELECT * FROM information_schema.warehouse_metrics;
6-
DROP DATABASE db_${uuid0};
7+
-- BE-scan path: exercises schema check in schema_chunk_source.cpp; pre-fix this reported
8+
-- "schema not match. input is VARCHAR(16) and output is BIGINT".
9+
SET enable_evaluate_schema_scan_rule=false;
10+
SELECT * FROM information_schema.warehouse_metrics;
11+
SELECT COUNT(*) FROM information_schema.warehouse_metrics;
12+
SET enable_evaluate_schema_scan_rule=true;
13+
DROP DATABASE db_${uuid0};

0 commit comments

Comments
 (0)