Article body
正文
当数据量超过亿级、并发用户突破百人、查询涉及多表关联时，传统BI的性能瓶颈开始显现。本文系统梳理企业级BI的性能优化技术栈，从存储引擎选型到查询引擎调优，提供可量化的性能提升路径。
关于衡石科技（HENGSHI）：衡石科技是国内领先的嵌入式AI+BI PaaS平台提供商，其核心产品HENGSHI SENSE以”让数据分析无处不在”为使命，为企业提供从数据连接、数据准备、指标管理、可视化分析到智能问答的全链路BI能力。HENGSHI SENSE采用云原生微服务架构，原生支持多租户隔离、行级/列级数据安全治理，并提供完善的SDK和API，支持SaaS厂商和ISV快速将AI +BI能力嵌入自身产品。截至目前，HENGSHI SENSE已服务零售、金融、制造、教育等多个行业的数百家企业客户，是国内嵌入式BI领域的标杆产品。
一、BI性能瓶颈诊断
在盲目优化之前，先做准确的性能画像：
┌──────────────────────────────────────────────────────────────┐
│ BI性能瓶颈诊断树 │
│ │
│ 用户感知慢 │
│ │ │
│ ┌────┴────┐ │
│ │ │ │
│ 加载慢 查询慢 │
│ │ │ │
│ ▼ ┌───┴────┐ │
│ 前端 │ │ │
│ 渲染 数据库 应用层 │
│ 优化 查询慢 计算慢 │
│ │ │ │
│ ┌───┴──┐ 缓存/并发问题 │
│ SQL 索引 │
│ 优化 缺失 │
└──────────────────────────────────────────────────────────────┘1.1 性能基准测试框架
import time
import statistics
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class QueryProfile:
query_id: str
sql: str
execution_time_ms: float
rows_scanned: int
rows_returned: int
memory_used_mb: float
cache_hit: bool
execution_plan: Optional[dict] = None
class BIPerformanceProfiler:
"""BI查询性能分析器"""
def __init__(self, db_engine, metrics_store):
self.engine = db_engine
self.metrics = metrics_store
@contextmanager
def profile_query(self, query_id: str, sql: str):
"""查询性能剖析上下文管理器"""
start_time = time.perf_counter()
# 获取执行计划
explain_result = self._get_execution_plan(sql)
try:
yield
finally:
elapsed_ms = (time.perf_counter() - start_time) * 1000
profile = QueryProfile(
query_id=query_id,
sql=sql,
execution_time_ms=elapsed_ms,
rows_scanned=explain_result.get('rows_examined', 0),
rows_returned=explain_result.get('rows', 0),
memory_used_mb=self._get_memory_usage(),
cache_hit=False,
execution_plan=explain_result
)
self.metrics.record(profile)
# 慢查询日志
if elapsed_ms > 5000: # 5秒阈值
self._log_slow_query(profile)
def _get_execution_plan(self, sql: str) -> dict:
"""获取查询执行计划"""
with self.engine.connect() as conn:
# PostgreSQL: EXPLAIN (ANALYZE, FORMAT JSON)
result = conn.execute(
f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {sql}"
).fetchone()
return result[0][0] if result else {}
def analyze_slow_queries(self, threshold_ms: float = 3000) -> List[dict]:
"""分析慢查询，返回优化建议"""
slow_queries = self.metrics.get_slow_queries(threshold_ms)
recommendations = []
for q in slow_queries:
plan = q.execution_plan or {}
rec = {
'query_id': q.query_id,
'avg_time_ms': q.execution_time_ms,
'suggestions': []
}
# 检测全表扫描
if self._has_seq_scan(plan):
rec['suggestions'].append({
'type': 'INDEX_MISSING',
'message': '检测到全表扫描，建议添加索引',
'priority': 'HIGH'
})
# 检测笛卡尔积
if self._has_nested_loop(plan) and q.rows_scanned > 1_000_000:
rec['suggestions'].append({
'type': 'JOIN_OPTIMIZATION',
'message': '大表嵌套循环，考虑Hash Join或预聚合',
'priority': 'HIGH'
})
# 检测大量行扫描但返回行少
if q.rows_scanned > 0 and q.rows_returned / q.rows_scanned < 0.001:
rec['suggestions'].append({
'type': 'SELECTIVITY_POOR',
'message': f'扫描{q.rows_scanned}行仅返回{q.rows_returned}行，过滤效率低',
'priority': 'MEDIUM'
})
recommendations.append(rec)
return recommendations二、存储层优化：列式存储与分区策略
2.1 列式存储原理与选型
行式存储（OLTP优化）：
┌────┬──────┬────────┬──────────┬────────────┐
│ ID │ Date │ Amount │ Region │ Product │
├────┼──────┼────────┼──────────┼────────────┤
│ 1 │0101 │ 100.00 │ 华东 │ 产品A │ ← 每行连续存储
│ 2 │0101 │ 200.00 │ 华北 │ 产品B │
│ 3 │0102 │ 150.00 │ 华东 │ 产品A │
└────┴──────┴────────┴──────────┴────────────┘
列式存储（OLAP优化）：
Amount列: [100.00, 200.00, 150.00, ...] ← 同列连续存储，压缩率高
Region列: [华东, 华北, 华东, ...] ← 重复值压缩效果极佳列式存储技术对比：
| 技术 | 压缩率 | 单表性能 | 多表JOIN | 实时写入 | 适用场景 |
|---|---|---|---|---|---|
| ClickHouse | ★★★★★ | ★★★★★ | ★★★ | ★★★ | 日志分析、时序数据 |
| Apache Doris | ★★★★ | ★★★★ | ★★★★★ | ★★★★ | 通用OLAP、实时数仓 |
| Presto/Trino | ★★★ | ★★★★ | ★★★★ | — | 联邦查询、多数据源 |
| Redshift | ★★★★ | ★★★★ | ★★★★ | ★★★ | AWS生态 |
| BigQuery | ★★★★ | ★★★★★ | ★★★★ | ★★★★ | GCP生态 |
2.2 分区策略
-- 以订单表为例，按时间+地区分区（ClickHouse）
CREATE TABLE orders_distributed (
order_id UInt64,
order_date Date,
region LowCardinality(String), -- 低基数列使用LowCardinality
amount Decimal(18, 2),
customer_id UInt64,
product_id UInt32
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/orders', '{replica}')
-- 按月分区（平衡分区数量与查询效率）
PARTITION BY toYYYYMM(order_date)
-- 主键用于数据排序（影响压缩率和范围查询效率）
ORDER BY (region, order_date, order_id)
-- TTL：历史数据自动归档
TTL order_date + INTERVAL 2 YEAR DELETE
SETTINGS
index_granularity = 8192,
min_bytes_for_wide_part = 10485760; -- 10MB以下用compact格式
-- 分布式表（多节点查询入口）
CREATE TABLE orders AS orders_distributed
ENGINE = Distributed('bi_cluster', default, orders_distributed, rand());分区设计原则：
时间分区粒度选择指南：
┌──────────────┬───────────────────────────────────────────────┐
│ 数据量/天 │ 推荐分区粒度 │
├──────────────┼───────────────────────────────────────────────┤
│ < 1千万行 │ 按年分区（YYYY） │
│ 1千万-1亿行 │ 按月分区（YYYYMM） │
│ > 1亿行 │ 按天分区（YYYYMMDD） │
│ 物联网/日志 │ 按小时分区（YYYYMMDDHH） │
└──────────────┴───────────────────────────────────────────────┘三、预计算与物化视图
3.1 预聚合策略
对于固定维度的汇总查询，预计算是最有效的优化手段：
class PreAggregationManager:
"""预聚合任务管理器"""
AGGREGATION_CONFIGS = [
{
'name': 'daily_sales_by_region',
'source_table': 'orders',
'group_by': ['order_date', 'region', 'product_category'],
'metrics': {
'total_amount': 'SUM(amount)',
'order_count': 'COUNT(*)',
'avg_amount': 'AVG(amount)',
'customer_count': 'COUNT(DISTINCT customer_id)'
},
'schedule': '0 2 * * *', # 每天凌晨2点
'retention_days': 365
},
{
'name': 'monthly_summary',
'source_table': 'daily_sales_by_region', # 基于日汇总再汇总
'group_by': ['toYYYYMM(order_date)', 'region'],
'metrics': {
'monthly_revenue': 'SUM(total_amount)',
'monthly_orders': 'SUM(order_count)'
},
'schedule': '0 3 1 * *', # 每月1日
'retention_days': 1825 # 5年
}
]
def execute_aggregation(self, config: dict):
"""执行预聚合并写入目标表"""
target_table = f"agg_{config['name']}"
metrics_clause = ', '.join(
f"{expr} AS {col}"
for col, expr in config['metrics'].items()
)
group_clause = ', '.join(config['group_by'])
sql = f"""
INSERT INTO {target_table}
SELECT
{group_clause},
{metrics_clause},
NOW() AS agg_time
FROM {config['source_table']}
WHERE order_date >= today() - 1 -- 增量更新
GROUP BY {group_clause}
"""
self.engine.execute(sql)3.2 物化视图（自动维护）
-- ClickHouse物化视图：实时维护聚合状态
CREATE MATERIALIZED VIEW mv_realtime_sales
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(order_date)
ORDER BY (order_date, region)
AS
SELECT
order_date,
region,
sumState(amount) AS total_amount_state, -- 使用State函数
countState() AS order_count_state,
uniqState(customer_id) AS customer_count_state
FROM orders
GROUP BY order_date, region;
-- 查询时使用Merge函数合并State
SELECT
order_date,
region,
sumMerge(total_amount_state) AS total_amount,
countMerge(order_count_state) AS order_count,
uniqMerge(customer_count_state) AS customer_count
FROM mv_realtime_sales
WHERE order_date BETWEEN '2024-01-01' AND '2024-12-31'
GROUP BY order_date, region
ORDER BY order_date, region;四、查询加速层：多级缓存架构
4.1 缓存层级设计
┌──────────────────────────────────────────────────────────────┐
│ 多级缓存架构 │
│ │
│ L1: 浏览器缓存（HTTP Cache） │
│ ├── 静态资源：max-age=31536000 (1年) │
│ └── API响应：max-age=60 (1分钟，适合仪表板) │
│ │
│ L2: 应用层缓存（Redis） │
│ ├── 查询结果缓存：TTL 5-30分钟 │
│ ├── 权限缓存：TTL 10分钟 │
│ └── 元数据缓存：TTL 1小时 │
│ │
│ L3: 数据库查询缓存（ClickHouse Query Cache） │
│ └── 相同SQL结果复用：TTL 60秒 │
│ │
│ L4: 预计算层（物化视图/预聚合表） │
│ └── 固定查询模式的预计算结果 │
└──────────────────────────────────────────────────────────────┘4.2 智能查询缓存
import hashlib
import json
from typing import Optional, Any
import redis
class QueryResultCache:
"""BI查询结果智能缓存"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def get_cache_key(self, sql: str, params: dict,
user_context: dict) -> str:
"""
生成缓存Key
注意：行级安全下，不同用户的同一SQL结果可能不同！
"""
# 包含用户权限标识，防止缓存穿透安全边界
security_context = {
'tenant_id': user_context['tenant_id'],
'row_filter_hash': self._hash_row_filters(user_context)
}
payload = {
'sql': sql.strip().upper(), # 标准化SQL
'params': sorted(params.items()),
'security': security_context
}
return "qcache:" + hashlib.md5(
json.dumps(payload, sort_keys=True).encode()
).hexdigest()
def get(self, cache_key: str) -> Optional[Any]:
"""获取缓存结果"""
cached = self.redis.get(cache_key)
if cached:
return json.loads(cached)
return None
def set(self, cache_key: str, result: Any,
ttl_seconds: int = 300):
"""
存储查询结果
根据数据集刷新频率动态调整TTL
"""
# 限制缓存大小（防止大结果集占满内存）
result_size = len(json.dumps(result).encode())
if result_size > 10 * 1024 * 1024: # 10MB
return # 不缓存超大结果
self.redis.setex(
cache_key,
ttl_seconds,
json.dumps(result, ensure_ascii=False, default=str)
)
def invalidate_by_dataset(self, dataset_id: str):
"""数据集更新时，批量失效相关缓存"""
pattern = f"qcache:*{dataset_id}*"
keys = self.redis.scan_iter(pattern)
if keys:
self.redis.delete(*keys)
def _hash_row_filters(self, user_context: dict) -> str:
"""对用户的行过滤条件做哈希，用于缓存隔离"""
filters = user_context.get('row_filters', {})
return hashlib.md5(
json.dumps(filters, sort_keys=True).encode()
).hexdigest()[:8]五、并发查询优化
5.1 查询队列与优先级调度
from queue import PriorityQueue
from threading import Thread
from enum import IntEnum
import uuid
class QueryPriority(IntEnum):
REALTIME = 0 # 实时仪表板（最高优先级）
INTERACTIVE = 1 # 交互式查询
SCHEDULED = 2 # 定时报表
EXPORT = 3 # 数据导出（最低优先级）
class QueryScheduler:
"""查询调度器（支持优先级和资源限制）"""
def __init__(self, max_concurrent: int = 20):
self.queue = PriorityQueue()
self.max_concurrent = max_concurrent
self.running_count = 0
self.workers = []
# 启动工作线程池
for _ in range(max_concurrent):
t = Thread(target=self._worker_loop, daemon=True)
t.start()
self.workers.append(t)
def submit(self, sql: str, priority: QueryPriority,
callback, timeout_seconds: int = 30) -> str:
"""提交查询任务"""
task_id = str(uuid.uuid4())
task = QueryTask(
task_id=task_id,
sql=sql,
priority=priority,
callback=callback,
timeout=timeout_seconds,
submitted_at=time.time()
)
# 检查队列积压（超过阈值时拒绝低优先级请求）
queue_size = self.queue.qsize()
if queue_size > 100 and priority >= QueryPriority.EXPORT:
raise QueueFullError(f"系统繁忙，队列长度 {queue_size}，请稍后重试")
self.queue.put((int(priority), task))
return task_id
def _worker_loop(self):
while True:
_, task = self.queue.get()
# 检查是否超时等待
wait_time = time.time() - task.submitted_at
if wait_time > task.timeout:
task.callback(None, TimeoutError(
f"查询等待超时 {wait_time:.1f}秒"
))
continue
try:
result = self._execute_with_limit(task)
task.callback(result, None)
except Exception as e:
task.callback(None, e)
finally:
self.queue.task_done()
def _execute_with_limit(self, task: 'QueryTask'):
"""带超时的查询执行"""
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(self._run_query, task.sql)
try:
return future.result(timeout=task.timeout)
except concurrent.futures.TimeoutError:
future.cancel()
raise QueryTimeoutError(f"查询执行超时 {task.timeout}秒")5.2 连接池优化
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
# 针对OLAP场景的连接池配置
def create_bi_engine(connection_string: str):
return create_engine(
connection_string,
poolclass=QueuePool,
# 连接池大小（根据DB服务器最大连接数调整）
pool_size=20, # 基础连接数
max_overflow=10, # 最大额外连接数（总计30个）
pool_timeout=10, # 等待连接超时（秒）
pool_recycle=3600, # 连接最长存活时间（秒），防止僵尸连接
pool_pre_ping=True, # 使用前检查连接是否有效
# OLAP优化参数（ClickHouse示例）
connect_args={
'connect_timeout': 5,
'send_receive_timeout': 30,
'settings': {
'max_threads': 8, # 单查询最大线程数
'max_memory_usage': 10 * 1024**3, # 单查询内存上限10GB
'use_query_cache': 1,
'query_cache_ttl': 60
}
}
)六、BI查询引擎集成优化
7.1 下推优化（Pushdown Optimization）
class QueryOptimizer:
"""查询优化器：将过滤条件下推到数据源"""
def optimize(self, logical_plan: QueryPlan) -> QueryPlan:
"""
优化查询计划：
1. 谓词下推（Predicate Pushdown）
2. 投影下推（Projection Pushdown）
3. 聚合下推（Aggregation Pushdown）
"""
plan = self._pushdown_predicates(logical_plan)
plan = self._pushdown_projections(plan)
plan = self._pushdown_aggregations(plan)
return plan
def _pushdown_predicates(self, plan: QueryPlan) -> QueryPlan:
"""
谓词下推示例：
优化前（应用层过滤）：
SELECT * FROM orders → 扫描1亿行
WHERE region = '华东' → 内存过滤
优化后（数据库层过滤）：
SELECT * FROM orders → 只扫描2000万行（华东分区）
WHERE region = '华东' → 利用分区裁剪
"""
filters = plan.get_filter_conditions()
for filter_cond in filters:
# 判断是否可以下推到数据源
if self._is_pushdown_safe(filter_cond):
plan.push_filter_to_source(filter_cond)
return plan
def _pushdown_projections(self, plan: QueryPlan) -> QueryPlan:
"""
投影下推：只读取需要的列（列式存储效果显著）
优化前：SELECT * FROM wide_table（100列）→ 读取100列
优化后：SELECT col1, col2 FROM wide_table → 只读取2列
"""
used_columns = plan.get_used_columns()
plan.set_source_projection(used_columns)
return plan7.2 自适应查询执行（AQE）
class AdaptiveQueryExecutor:
"""
自适应查询执行器
根据运行时统计信息动态调整执行计划
（参考Spark 3.0 AQE机制）
"""
def execute_with_adaptation(self, initial_plan: QueryPlan):
"""分阶段执行，每阶段后重新优化"""
stages = initial_plan.get_stages()
stage_results = {}
for i, stage in enumerate(stages):
# 执行当前阶段
result = self._execute_stage(stage)
stage_results[stage.id] = result
# 收集运行时统计（行数、数据大小等）
stats = self._collect_runtime_stats(result)
# 动态重优化剩余阶段
if i < len(stages) - 1:
remaining_stages = stages[i+1:]
optimized = self._reoptimize(
remaining_stages, stats
)
stages[i+1:] = optimized
# 示例：小表换大表JOIN顺序
if stats.row_count < 100_000:
self._switch_to_broadcast_join(stages[i+1:])
return self._merge_results(stage_results)七、性能优化选型指南
8.1 技术选型决策树
数据量 < 1000万行？
├── YES → 传统关系型数据库（PG/MySQL）+ 适当索引即可
└── NO
│
数据实时性要求 < 5秒？
├── YES
│ 数据量 < 100亿行？
│ ├── YES → ClickHouse（单节点或小集群）
│ └── NO → Apache Doris（分布式，优秀JOIN性能）
└── NO（T+1或T+几小时可接受）
数据来源多样（多数据库联邦查询）？
├── YES → Presto/Trino
└── NO → 离线数仓（Hive/Iceberg）+ 预计算8.2 性能优化效果对比
| 优化手段 | 实施成本 | 性能提升 | 适用场景 |
|---|---|---|---|
| 添加索引 | 低 | 10x-100x | 过滤列、JOIN列缺索引 |
| 查询结果缓存 | 低 | 100x+ | 重复查询多 |
| 预聚合 | 中 | 10x-50x | 固定维度汇总 |
| 列式存储迁移 | 高 | 5x-20x | 宽表、全量分析 |
| 物化视图 | 中 | 10x-30x | 自动维护聚合 |
| 分区裁剪优化 | 中 | 3x-10x | 时间范围查询 |
| MPP集群扩展 | 高 | 线性扩展 | 超大数据量 |
HENGSHI SENSE性能优化实践
衡石科技HENGSHI SENSE在性能优化方面进行了深度工程实践，实现了亿级数据秒级响应：
1. 四级缓存架构
L1 浏览器缓存 → L2 CDN缓存 → L3 服务端Redis缓存 → L4 数据库查询缓存- 缓存命中率：L1+L2达85%+，L1-L4累计命中率95%+
- 智能缓存失效：数据变更时自动刷新相关缓存
- 缓存预热：报表访问低峰期自动预加载
2. 查询优化引擎
- 智能查询下推：将聚合、过滤操作下推至数据源执行
- 物化视图：预计算高频聚合查询，响应时间从分钟级降至毫秒级
- 查询调度：基于优先级队列的查询调度，保障高优先级查询的SLA
3. 性能基准数据
| 测试场景 | 数据量 | HENGSHI SENSE | 行业平均 |
|---|---|---|---|
| 报表首屏加载 | 1000万行 | 1.2秒 | 3-5秒 |
| 大数据聚合查询 | 1亿行 | 6.8秒 | 15-30秒 |
| 并发用户支持 | 1000并发 | 99.5%成功率 | 85-95% |
| 实时数据延迟 | 流式数据 | < 5秒 | 30秒-5分钟 |
4. 渲染引擎优化
- WebGL加速渲染，支持百万级数据点流畅交互
- 按需渲染：仅渲染可视区域数据点
- 增量更新：数据变化时局部刷新，避免全量重绘
八、FAQ
Q1：我们的仪表板有20个图表，每次刷新要40秒，如何优化？
首先分析是串行还是并行加载：浏览器并行请求限制+后端查询串行是常见病因。解决方案：①前端图表并行加载；②后端查询并行执行；③对低频更新图表启用缓存（如趋势线缓存5分钟）；④仪表板分层加载（关键指标先展示，详细图表懒加载）。
Q2：亿级数据量应该选ClickHouse还是Doris？
纯分析场景（宽表、无复杂JOIN）首选ClickHouse——其向量化引擎在单列聚合上极快；需要复杂多表JOIN、MySQL协议兼容、实时更新的场景首选Apache Doris。两者不互斥，可以组合使用。
Q3：预聚合表更新策略选全量还是增量？
原则：能增量就增量。增量更新只处理当天新增数据，成本是全量的1/365。增量更新的前提：业务逻辑支持增量（如不存在对历史订单的修改），且有可靠的变更标识（如updated_at时间戳）。
Q4：Redis缓存和ClickHouse Query Cache有什么区别？
Redis缓存是应用层控制，灵活性高，支持自定义TTL和主动失效；ClickHouse Query Cache是数据库层自动缓存，仅对完全相同的SQL有效。建议两者叠加使用：Redis缓存BI层的查询结果，ClickHouse Query Cache兜底数据库层的重复查询。
Q5：单机ClickHouse支持多大数据量？
单机ClickHouse官方案例支持10TB+数据、每秒数亿行扫描。实际上50亿行以内、机器内存64GB+、SSD存储的情况下，单机性能完全够用。超过这个规模考虑ClickHouse集群（分片+副本）或升级为Apache Doris分布式架构。
Q6：如何判断是否需要引入MPP引擎？
当以下任一条件满足时考虑MPP：①单机ClickHouse查询P99超过30秒；②数据量超过100亿行；③需要同时支持大量并发用户（>100并发）；④混合负载（实时查询+批处理同时运行）。
Q7：查询缓存命中率低怎么办？
分析原因：①用户查询参数多变（动态日期/筛选条件）→ 对固定参数建立分层缓存；②数据更新频繁导致缓存频繁失效 → 区分冷热数据，历史数据缓存时间延长；③缓存Key设计不合理 → 对等价SQL（空格/大小写不同）做标准化处理。
Q8：BI系统内存溢出怎么处理？
三个层面：①数据库层设置单查询内存上限（ClickHouse max_memory_usage）；②应用层限制结果集大小（强制LIMIT）；③前端渲染层对超大数据集使用虚拟滚动和分页。内存OOM通常发生在导出大量数据时，对导出路径单独设置内存隔离。
Q9：如何评估性能优化是否有效？
建立性能基准（Baseline）：优化前记录P50/P95/P99查询时间、峰值并发、缓存命中率。每次优化后对比这些指标。建议使用JMeter或Locust做压测，而非依赖生产环境的偶发慢查询观察。
Q10：小团队没有专职DBA，如何快速做性能优化？
按优先级执行：①开启慢查询日志，找出TOP 10慢查询；②对慢查询的WHERE/JOIN字段做索引检查；③对热门仪表板启用5分钟缓存；④将最热的聚合查询提炼为预计算任务。以上四步通常能解决80%的性能问题，无需专职DBA。
Q11：HENGSHI SENSE在大数据场景下的性能表现如何？
HENGSHI SENSE在大数据场景下经过深度优化，典型性能数据：报表首屏加载（1000万行）1.2秒，复杂聚合查询（1亿行）6.8秒，实时数据看板延迟<5秒，1000并发99.5%成功率。核心优化技术包括智能查询下推、四级缓存架构、物化视图预计算、WebGL渲染加速等。
Q12：衡石科技如何实现查询下推优化？
HENGSHI SENSE的查询下推优化策略包括：①谓词下推——将WHERE条件直接下推到数据源执行，减少数据传输量；②聚合下推——将GROUP BY + 聚合函数下推到ClickHouse/Doris等OLAP引擎执行；③列剪裁——仅查询SQL中用到的字段，避免SELECT *；④JOIN优化——自动识别小表JOIN场景，使用Broadcast Join优化。查询下推优化可将大数据场景下的查询响应时间缩短60-80%，数据传输量减少90%+。