⚡️ perf(export): 重构大结果集导出链路并支持流式写入

- 新增 ExportFileOptions 统一承载导出格式、进度任务和 XLSX sheet 行数上限
- 查询导出改为流式写入文件，避免一次性缓存整批结果导致高内存占用
- 增加值数组快速路径并复用扫描与写入缓冲，减少逐行 map 分配开销
- 为 ClickHouse、自定义驱动、达梦、SQLServer 和 TDengine 补齐 StreamQuery 支持
- 导出时间字符串仅在形似时间时再解析，避免普通文本被误判改写
- 补充 XLSX 分 sheet、流式导出和基准测试覆盖

internal/db/clickhouse_impl.go

@@ -777,6 +777,22 @@ func (c *ClickHouseDB) Query(query string) ([]map[string]interface{}, []string,
 	return scanRows(rows)
 }
 
+func (c *ClickHouseDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if c.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+	rows, err := c.conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	return streamRows(rows, consumer)
+}
+
+func (c *ClickHouseDB) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return c.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (c *ClickHouseDB) ExecContext(ctx context.Context, query string) (int64, error) {
 	if c.conn == nil {
 		return 0, fmt.Errorf("连接未打开")

internal/db/custom_impl.go

@@ -111,6 +111,24 @@ func (c *CustomDB) Query(query string) ([]map[string]interface{}, []string, erro
 	return scanRowsForDialect(rows, c.scanDialect())
 }
 
+func (c *CustomDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if c.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+
+	rows, err := c.conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	return streamRowsForDialect(rows, c.scanDialect(), consumer)
+}
+
+func (c *CustomDB) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return c.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (c *CustomDB) scanDialect() string {
 	if strings.EqualFold(strings.TrimSpace(c.driver), "mysql") {
 		return "mysql"

internal/db/dameng_impl.go

@@ -182,6 +182,24 @@ func (d *DamengDB) Query(query string) ([]map[string]interface{}, []string, erro
 	return scanRows(rows)
 }
 
+func (d *DamengDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if d.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+
+	rows, err := d.conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	return streamRows(rows, consumer)
+}
+
+func (d *DamengDB) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return d.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (d *DamengDB) ExecContext(ctx context.Context, query string) (int64, error) {
 	if d.conn == nil {
 		return 0, fmt.Errorf("连接未打开")

internal/db/database.go

@@ -76,6 +76,28 @@ type StatementQueryExecer interface {
 	QueryContext(ctx context.Context, query string) ([]map[string]interface{}, []string, error)
 }
 
+// QueryStreamConsumer receives query metadata and rows incrementally.
+// Implementations can stream rows directly to files to avoid buffering entire result sets in memory.
+type QueryStreamConsumer interface {
+	SetColumns(columns []string) error
+	ConsumeRow(row map[string]interface{}) error
+}
+
+// QueryStreamValueConsumer is an optional fast path for stream consumers that
+// can consume normalized row values in column order without requiring a
+// map[string]interface{} allocation per row.
+type QueryStreamValueConsumer interface {
+	SetColumns(columns []string) error
+	ConsumeRowValues(values []interface{}) error
+}
+
+// StreamQueryExecer is an optional interface for drivers or pinned sessions that can
+// stream query rows incrementally instead of materializing []map rows in memory.
+type StreamQueryExecer interface {
+	StreamQuery(query string, consumer QueryStreamConsumer) error
+	StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error
+}
+
 // StatementQueryMessageExecer can run queries on a pinned session and return
 // extra server messages/notices alongside rows.
 type StatementQueryMessageExecer interface {
@@ -178,6 +200,22 @@ func (e *sqlConnStatementExecer) Query(query string) ([]map[string]interface{},
 	return e.QueryContext(context.Background(), query)
 }
 
+func (e *sqlConnStatementExecer) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if e == nil || e.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+	rows, err := e.conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	return streamRowsForDialect(rows, e.scanDialect, consumer)
+}
+
+func (e *sqlConnStatementExecer) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return e.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (e *sqlConnStatementExecer) QueryMultiContext(ctx context.Context, query string) ([]connection.ResultSetData, error) {
 	if e == nil || e.conn == nil {
 		return nil, fmt.Errorf("连接未打开")
@@ -275,6 +313,23 @@ func (e *sqlConnTransactionExecer) Query(query string) ([]map[string]interface{}
 	return e.QueryContext(context.Background(), query)
 }
 
+func (e *sqlConnTransactionExecer) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	conn, err := e.activeConn()
+	if err != nil {
+		return err
+	}
+	rows, err := conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	return streamRowsForDialect(rows, e.scanDialect, consumer)
+}
+
+func (e *sqlConnTransactionExecer) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return e.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (e *sqlConnTransactionExecer) QueryMultiContext(ctx context.Context, query string) ([]connection.ResultSetData, error) {
 	conn, err := e.activeConn()
 	if err != nil {
@@ -401,6 +456,23 @@ func (e *sqlTxStatementExecer) Query(query string) ([]map[string]interface{}, []
 	return e.QueryContext(context.Background(), query)
 }
 
+func (e *sqlTxStatementExecer) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	tx, err := e.activeTx()
+	if err != nil {
+		return err
+	}
+	rows, err := tx.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	return streamRows(rows, consumer)
+}
+
+func (e *sqlTxStatementExecer) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return e.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (e *sqlTxStatementExecer) QueryMultiContext(ctx context.Context, query string) ([]connection.ResultSetData, error) {
 	tx, err := e.activeTx()
 	if err != nil {

internal/db/scan_rows.go

@@ -11,6 +11,19 @@ func scanRows(rows *sql.Rows) ([]map[string]interface{}, []string, error) {
 	return scanRowsForDialect(rows, "")
 }
 
+func streamRows(rows *sql.Rows, consumer QueryStreamConsumer) error {
+	return streamRowsForDialect(rows, "", consumer)
+}
+
+type queryRowScanner struct {
+	columns     []string
+	dbTypeNames []string
+	dialect     string
+	values      []interface{}
+	normalized  []interface{}
+	valuePtrs   []interface{}
+}
+
 func scanRowsForDialect(rows *sql.Rows, dialect string) ([]map[string]interface{}, []string, error) {
 	columns, err := rows.Columns()
 	if err != nil {
@@ -23,27 +36,14 @@ func scanRowsForDialect(rows *sql.Rows, dialect string) ([]map[string]interface{
 		colTypes = nil
 	}
 
+	scanner := newQueryRowScanner(columns, colTypes, dialect)
 	resultData := make([]map[string]interface{}, 0)
 
 	for rows.Next() {
-		values := make([]interface{}, len(columns))
-		valuePtrs := make([]interface{}, len(columns))
-		for i := range columns {
-			valuePtrs[i] = &values[i]
-		}
-
-		if err := rows.Scan(valuePtrs...); err != nil {
+		entry, err := scanner.scanCurrentRow(rows)
+		if err != nil {
 			continue
 		}
-
-		entry := make(map[string]interface{}, len(columns))
-		for i, col := range columns {
-			dbTypeName := ""
-			if colTypes != nil && i < len(colTypes) && colTypes[i] != nil {
-				dbTypeName = colTypes[i].DatabaseTypeName()
-			}
-			entry[col] = normalizeQueryValueWithDBTypeAndDialect(values[i], dbTypeName, dialect)
-		}
 		resultData = append(resultData, entry)
 	}
 
@@ -53,6 +53,95 @@ func scanRowsForDialect(rows *sql.Rows, dialect string) ([]map[string]interface{
 	return resultData, columns, nil
 }
 
+func streamRowsForDialect(rows *sql.Rows, dialect string, consumer QueryStreamConsumer) error {
+	if consumer == nil {
+		return fmt.Errorf("query stream consumer required")
+	}
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return err
+	}
+	columns = ensureUniqueQueryColumnNames(columns)
+
+	colTypes, err := rows.ColumnTypes()
+	if err != nil || len(colTypes) != len(columns) {
+		colTypes = nil
+	}
+
+	scanner := newQueryRowScanner(columns, colTypes, dialect)
+	if err := consumer.SetColumns(columns); err != nil {
+		return err
+	}
+	valueConsumer, useValueConsumer := consumer.(QueryStreamValueConsumer)
+
+	for rows.Next() {
+		if useValueConsumer {
+			values, err := scanner.scanCurrentRowValues(rows)
+			if err != nil {
+				continue
+			}
+			if err := valueConsumer.ConsumeRowValues(values); err != nil {
+				return err
+			}
+			continue
+		}
+		entry, err := scanner.scanCurrentRow(rows)
+		if err != nil {
+			continue
+		}
+		if err := consumer.ConsumeRow(entry); err != nil {
+			return err
+		}
+	}
+
+	return rows.Err()
+}
+
+func newQueryRowScanner(columns []string, colTypes []*sql.ColumnType, dialect string) *queryRowScanner {
+	values := make([]interface{}, len(columns))
+	valuePtrs := make([]interface{}, len(columns))
+	for i := range columns {
+		valuePtrs[i] = &values[i]
+	}
+	dbTypeNames := make([]string, len(columns))
+	for i := range columns {
+		if colTypes != nil && i < len(colTypes) && colTypes[i] != nil {
+			dbTypeNames[i] = colTypes[i].DatabaseTypeName()
+		}
+	}
+	return &queryRowScanner{
+		columns:     columns,
+		dbTypeNames: dbTypeNames,
+		dialect:     dialect,
+		values:      values,
+		normalized:  make([]interface{}, len(columns)),
+		valuePtrs:   valuePtrs,
+	}
+}
+
+func (s *queryRowScanner) scanCurrentRowValues(rows *sql.Rows) ([]interface{}, error) {
+	if err := rows.Scan(s.valuePtrs...); err != nil {
+		return nil, err
+	}
+	for i := range s.columns {
+		s.normalized[i] = normalizeQueryValueWithDBTypeAndDialect(s.values[i], s.dbTypeNames[i], s.dialect)
+	}
+	return s.normalized, nil
+}
+
+func (s *queryRowScanner) scanCurrentRow(rows *sql.Rows) (map[string]interface{}, error) {
+	normalized, err := s.scanCurrentRowValues(rows)
+	if err != nil {
+		return nil, err
+	}
+	entry := make(map[string]interface{}, len(s.columns))
+	for i, col := range s.columns {
+		entry[col] = normalized[i]
+	}
+	return entry, nil
+}
+
 func ensureUniqueQueryColumnNames(columns []string) []string {
 	if len(columns) == 0 {
 		return columns

internal/db/sqlserver_impl.go

@@ -385,6 +385,23 @@ func (e *sqlServerSessionExecer) QueryContext(ctx context.Context, query string)
 	return rows, columns, err
 }
 
+func (e *sqlServerSessionExecer) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if e == nil || e.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+	retmsg := &sqlexp.ReturnMessage{}
+	rows, err := e.conn.QueryContext(ctx, query, retmsg)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	return streamRows(rows, consumer)
+}
+
+func (e *sqlServerSessionExecer) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return e.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (e *sqlServerSessionExecer) QueryWithMessages(query string) ([]map[string]interface{}, []string, []string, error) {
 	return e.QueryContextWithMessages(context.Background(), query)
 }

internal/db/tdengine_impl.go

@@ -168,6 +168,24 @@ func (t *TDengineDB) Query(query string) ([]map[string]interface{}, []string, er
 	return scanRows(rows)
 }
 
+func (t *TDengineDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if t.conn == nil {
+		return fmt.Errorf("连接未打开")
+	}
+
+	rows, err := t.conn.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	return streamRows(rows, consumer)
+}
+
+func (t *TDengineDB) StreamQuery(query string, consumer QueryStreamConsumer) error {
+	return t.StreamQueryContext(context.Background(), query, consumer)
+}
+
 func (t *TDengineDB) ExecContext(ctx context.Context, query string) (int64, error) {
 	if t.conn == nil {
 		return 0, fmt.Errorf("连接未打开")