🐛 fix(messaging): 避免字段元数据读取样本消息 (#913)

## 背景与问题 Kafka、MQTT 和 RocketMQ 的 `GetColumns` 会读取最多 20 条实际消息并推断动态字段。查看字段元数据因此可能产生等待、订阅或 Broker 消费活动，并要求额外消息读取权限。 ## 变更点 - Kafka、MQTT 和 RocketMQ 的 `GetColumns` 只返回静态消息字段，不再调用 `FetchMessages` - `GetAllColumns` 继续复用静态 `GetColumns`，批量字段加载也不会读取消息 - 保留显式 `SELECT` 查询消息及其动态结果字段行为 - 为三类消息服务补充回归断言，验证字段元数据读取次数为 0，并精确校验静态字段集合 ## 影响范围 - 字段元数据与自动补全不再展示从样本消息推断出的动态字段 - 实际查询结果仍会根据消息内容返回动态字段 - 不影响显式消息查询、发布及 Topic 元数据能力 ## 验证 - `go test ./internal/db -run "Test(KafkaGetColumnsReturnsStaticFieldsWithoutFetchingMessages|MQTTQueryExecAndColumns|RocketMQQueryExecAndColumns)$" -count=1` - `go test ./internal/db -count=1` - 独立对抗性代码审查通过；审查发现的字段断言子串误匹配已改为精确匹配 Closes #898
2026-08-15 11:14:31 +08:00 · 2026-08-11 09:05:12 +08:00
parent 3c77ad0bb3 93fa4a7630
commit 3adedff09c
6 changed files with 36 additions and 109 deletions
--- a/internal/db/kafka_impl.go
+++ b/internal/db/kafka_impl.go
@@ -335,17 +335,6 @@ func (k *KafkaDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefin
 	if topic == "" {
 		return nil, fmt.Errorf("Kafka topic 不能为空")
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	records, err := k.runtime.FetchMessages(ctx, kafkaFetchRequest{
-		Topic:  topic,
-		Limit:  20,
-		Latest: false,
-	})
-	if err != nil {
-		return nil, err
-	}
-	rows := kafkaMessageRows(records)
 	columns := []connection.ColumnDefinition{
 		{Name: "topic", Type: "string", Nullable: "NO", Comment: "Kafka topic"},
 		{Name: "partition", Type: "int", Nullable: "NO", Key: "PRI", Comment: "Kafka partition id"},
@@ -358,27 +347,6 @@ func (k *KafkaDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefin
 		{Name: "key_size", Type: "int", Nullable: "YES", Comment: "Message key size in bytes"},
 		{Name: "value_size", Type: "int", Nullable: "YES", Comment: "Message value size in bytes"},
 	}
-	seen := map[string]struct{}{
-		"topic": {}, "partition": {}, "offset": {}, "timestamp": {}, "high_water_mark": {},
-		"key": {}, "value": {}, "headers": {}, "key_size": {}, "value_size": {},
-	}
-	for _, row := range rows {
-		for key, value := range row {
-			if _, exists := seen[key]; exists {
-				continue
-			}
-			if !strings.HasPrefix(key, "headers.") && !strings.HasPrefix(key, "key.") && !strings.HasPrefix(key, "value.") {
-				continue
-			}
-			seen[key] = struct{}{}
-			columns = append(columns, connection.ColumnDefinition{
-				Name:     key,
-				Type:     inferChromaValueType(value),
-				Nullable: "YES",
-				Comment:  "Derived Kafka field",
-			})
-		}
-	}
 	return columns, nil
 }

--- a/internal/db/kafka_impl_test.go
+++ b/internal/db/kafka_impl_test.go
@@ -20,6 +20,7 @@ type fakeKafkaRuntime struct {
 	lastDescribeTopic  string
 	lastFetchRequest   kafkaFetchRequest
 	lastPublishCommand kafkaPublishCommand
+	fetchCount         int
 }

 func TestKafkaRuntimeDoesNotDeriveRequestTimeoutFromConnectionTimeout(t *testing.T) {
@@ -84,6 +85,7 @@ func (f *fakeKafkaRuntime) DescribeTopic(ctx context.Context, topic string) (kaf
 }

 func (f *fakeKafkaRuntime) FetchMessages(ctx context.Context, request kafkaFetchRequest) ([]kafkaMessageRecord, error) {
+	f.fetchCount++
 	f.lastFetchRequest = request
 	return append([]kafkaMessageRecord(nil), f.fetchResult...), nil
 }
@@ -250,7 +252,7 @@ func TestKafkaExecPublishesJSONCommand(t *testing.T) {
 	}
 }

-func TestKafkaGetColumnsIncludesDerivedFields(t *testing.T) {
+func TestKafkaGetColumnsReturnsStaticFieldsWithoutFetchingMessages(t *testing.T) {
 	runtime := &fakeKafkaRuntime{
 		fetchResult: []kafkaMessageRecord{{
 			Message: kafka.Message{Topic: "orders.events"},
@@ -268,14 +270,22 @@ func TestKafkaGetColumnsIncludesDerivedFields(t *testing.T) {
 	if err != nil {
 		t.Fatalf("GetColumns failed: %v", err)
 	}
+	if runtime.fetchCount != 0 {
+		t.Fatalf("GetColumns must not read Kafka messages, fetches=%d", runtime.fetchCount)
+	}
 	names := make([]string, 0, len(columns))
 	for _, col := range columns {
 		names = append(names, col.Name)
 	}
 	joined := strings.Join(names, ",")
-	for _, want := range []string{"topic", "partition", "offset", "value.meta.ip", "headers.x-request-id"} {
-		if !strings.Contains(joined, want) {
-			t.Fatalf("expected derived column %q in %s", want, joined)
+	for _, want := range []string{"topic", "partition", "offset", "value", "headers"} {
+		if !containsString(names, want) {
+			t.Fatalf("expected Kafka column %q in %s", want, joined)
+		}
+	}
+	for _, unexpected := range []string{"value.meta.ip", "headers.x-request-id"} {
+		if containsString(names, unexpected) {
+			t.Fatalf("unexpected sample-derived Kafka column %q in %s", unexpected, joined)
 		}
 	}
 }
--- a/internal/db/mqtt_impl.go
+++ b/internal/db/mqtt_impl.go
@@ -337,18 +337,6 @@ func (m *MQTTDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefini
 	if topic == "" {
 		return nil, fmt.Errorf("MQTT topic 不能为空")
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	records, err := m.runtime.FetchMessages(ctx, mqttFetchRequest{
-		Topic: topic,
-		Limit: 20,
-		QoS:   m.defaultQoS,
-		Wait:  m.fetchWait,
-	})
-	if err != nil {
-		return nil, err
-	}
-	rows := mqttMessageRows(records)
 	columns := []connection.ColumnDefinition{
 		{Name: "topic", Type: "string", Nullable: "NO", Comment: "MQTT topic"},
 		{Name: "qos", Type: "tinyint", Nullable: "NO", Comment: "MQTT QoS level"},
@@ -360,27 +348,6 @@ func (m *MQTTDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefini
 		{Name: "payload_bytes", Type: "int", Nullable: "YES", Comment: "Payload size in bytes"},
 		{Name: "received_at", Type: "timestamp", Nullable: "YES", Comment: "Client receive timestamp"},
 	}
-	seen := map[string]struct{}{
-		"topic": {}, "qos": {}, "retained": {}, "duplicate": {}, "message_id": {},
-		"payload": {}, "payload_encoding": {}, "payload_bytes": {}, "received_at": {},
-	}
-	for _, row := range rows {
-		for key, value := range row {
-			if _, exists := seen[key]; exists {
-				continue
-			}
-			if !strings.HasPrefix(key, "payload.") {
-				continue
-			}
-			seen[key] = struct{}{}
-			columns = append(columns, connection.ColumnDefinition{
-				Name:     key,
-				Type:     inferChromaValueType(value),
-				Nullable: "YES",
-				Comment:  "Derived MQTT payload field",
-			})
-		}
-	}
 	return columns, nil
 }

--- a/internal/db/mqtt_impl_test.go
+++ b/internal/db/mqtt_impl_test.go
@@ -181,20 +181,27 @@ func TestMQTTQueryExecAndColumns(t *testing.T) {
 		t.Fatalf("unexpected mqtt publish command: %#v", fakeRuntime.published[0])
 	}

+	fakeRuntime.fetchRequests = nil
 	columnDefs, err := client.GetColumns(mqttSyntheticDatabase, "devices/+/telemetry")
 	if err != nil {
 		t.Fatalf("GetColumns failed: %v", err)
 	}
+	if len(fakeRuntime.fetchRequests) != 0 {
+		t.Fatalf("GetColumns must not read MQTT messages, requests=%d", len(fakeRuntime.fetchRequests))
+	}
 	names := make([]string, 0, len(columnDefs))
 	for _, col := range columnDefs {
 		names = append(names, col.Name)
 	}
 	joined := strings.Join(names, ",")
-	for _, want := range []string{"topic", "payload.meta.source", "payload_encoding"} {
-		if !strings.Contains(joined, want) {
+	for _, want := range []string{"topic", "payload", "payload_encoding"} {
+		if !containsString(names, want) {
 			t.Fatalf("expected mqtt column %q in %s", want, joined)
 		}
 	}
+	if containsString(names, "payload.meta.source") {
+		t.Fatalf("unexpected sample-derived MQTT column in %s", joined)
+	}

 	databases, err := client.GetDatabases()
 	if err != nil {
--- a/internal/db/rocketmq_impl.go
+++ b/internal/db/rocketmq_impl.go
@@ -390,20 +390,6 @@ func (r *RocketMQDB) GetColumns(dbName, tableName string) ([]connection.ColumnDe
 	if topic == "" {
 		return nil, fmt.Errorf("RocketMQ topic 不能为空")
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	records, err := r.runtime.FetchMessages(ctx, rocketmqFetchRequest{
-		Topic:         topic,
-		Limit:         20,
-		ConsumerGroup: r.resolveConsumerGroup("columns"),
-		TagExpression: r.defaultTagExpression,
-		Latest:        false,
-		PullBatchSize: r.pullBatchSize,
-	})
-	if err != nil {
-		return nil, err
-	}
-	rows := rocketmqMessageRows(records)
 	columns := []connection.ColumnDefinition{
 		{Name: "topic", Type: "string", Nullable: "NO", Comment: "RocketMQ topic"},
 		{Name: "broker_name", Type: "string", Nullable: "NO", Comment: "Broker name"},
@@ -420,28 +406,6 @@ func (r *RocketMQDB) GetColumns(dbName, tableName string) ([]connection.ColumnDe
 		{Name: "body_encoding", Type: "string", Nullable: "YES", Comment: "Message body encoding"},
 		{Name: "properties", Type: "json", Nullable: "YES", Comment: "Message properties"},
 	}
-	seen := map[string]struct{}{
-		"topic": {}, "broker_name": {}, "queue_id": {}, "queue_offset": {}, "msg_id": {}, "offset_msg_id": {},
-		"tags": {}, "keys": {}, "born_timestamp": {}, "store_timestamp": {}, "reconsume_times": {},
-		"body": {}, "body_encoding": {}, "properties": {},
-	}
-	for _, row := range rows {
-		for key, value := range row {
-			if _, exists := seen[key]; exists {
-				continue
-			}
-			if !strings.HasPrefix(key, "body.") && !strings.HasPrefix(key, "properties.") {
-				continue
-			}
-			seen[key] = struct{}{}
-			columns = append(columns, connection.ColumnDefinition{
-				Name:     key,
-				Type:     inferChromaValueType(value),
-				Nullable: "YES",
-				Comment:  "Derived RocketMQ field",
-			})
-		}
-	}
 	return columns, nil
 }

--- a/internal/db/rocketmq_impl_test.go
+++ b/internal/db/rocketmq_impl_test.go
@@ -21,6 +21,7 @@ type fakeRocketMQRuntime struct {
 	lastDescribe       rocketmqDescribeRequest
 	lastFetch          rocketmqFetchRequest
 	lastPublish        rocketmqPublishCommand
+	fetchCount         int
 }

 func (f *fakeRocketMQRuntime) Close() error { return nil }
@@ -44,6 +45,7 @@ func (f *fakeRocketMQRuntime) DescribeTopic(ctx context.Context, request rocketm
 }

 func (f *fakeRocketMQRuntime) FetchMessages(ctx context.Context, request rocketmqFetchRequest) ([]rocketmqMessageRecord, error) {
+	f.fetchCount++
 	f.lastFetch = request
 	items := append([]rocketmqMessageRecord(nil), f.fetchResult...)
 	if request.Offset > 0 {
@@ -314,20 +316,29 @@ func TestRocketMQQueryExecAndColumns(t *testing.T) {
 		t.Fatalf("unexpected publish properties: %#v", fakeRuntime.lastPublish.Properties)
 	}

+	fakeRuntime.fetchCount = 0
 	columnDefs, err := client.GetColumns(rocketMQSyntheticDatabase, "orders.events")
 	if err != nil {
 		t.Fatalf("GetColumns failed: %v", err)
 	}
+	if fakeRuntime.fetchCount != 0 {
+		t.Fatalf("GetColumns must not read RocketMQ messages, fetches=%d", fakeRuntime.fetchCount)
+	}
 	names := make([]string, 0, len(columnDefs))
 	for _, col := range columnDefs {
 		names = append(names, col.Name)
 	}
 	joined := strings.Join(names, ",")
-	for _, want := range []string{"topic", "body.meta.source", "properties.trace"} {
-		if !strings.Contains(joined, want) {
+	for _, want := range []string{"topic", "body", "properties"} {
+		if !containsString(names, want) {
 			t.Fatalf("expected rocketmq column %q in %s", want, joined)
 		}
 	}
+	for _, unexpected := range []string{"body.meta.source", "properties.trace"} {
+		if containsString(names, unexpected) {
+			t.Fatalf("unexpected sample-derived RocketMQ column %q in %s", unexpected, joined)
+		}
+	}

 	databases, err := client.GetDatabases()
 	if err != nil {