diff --git a/Cargo.lock b/Cargo.lock
index a09fdec..291dd8a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1251,6 +1251,15 @@ dependencies = [
  "wasmtime-wasi",
 ]
 
+[[package]]
+name = "erp-plugin-crm"
+version = "0.1.0"
+dependencies = [
+ "serde",
+ "serde_json",
+ "wit-bindgen 0.55.0",
+]
+
 [[package]]
 name = "erp-plugin-prototype"
 version = "0.1.0"
diff --git a/crates/erp-plugin/src/data_dto.rs b/crates/erp-plugin/src/data_dto.rs
index 4b76114..42a2337 100644
--- a/crates/erp-plugin/src/data_dto.rs
+++ b/crates/erp-plugin/src/data_dto.rs
@@ -36,3 +36,30 @@ pub struct PluginDataListParams {
     /// "asc" or "desc"
     pub sort_order: Option<String>,
 }
+
+/// 聚合查询响应项
+#[derive(Debug, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct AggregateItem {
+    /// 分组键（字段值）
+    pub key: String,
+    /// 计数
+    pub count: i64,
+}
+
+/// 聚合查询参数
+#[derive(Debug, Serialize, Deserialize, utoipa::IntoParams)]
+pub struct AggregateQueryParams {
+    /// 分组字段名
+    pub group_by: String,
+    /// JSON 格式过滤: {"field":"value"}
+    pub filter: Option<String>,
+}
+
+/// 统计查询参数
+#[derive(Debug, Serialize, Deserialize, utoipa::IntoParams)]
+pub struct CountQueryParams {
+    /// 搜索关键词
+    pub search: Option<String>,
+    /// JSON 格式过滤: {"field":"value"}
+    pub filter: Option<String>,
+}
diff --git a/crates/erp-plugin/src/data_service.rs b/crates/erp-plugin/src/data_service.rs
index 6568287..e5c1406 100644
--- a/crates/erp-plugin/src/data_service.rs
+++ b/crates/erp-plugin/src/data_service.rs
@@ -261,6 +261,99 @@ impl PluginDataService {
 
         Ok(())
     }
+
+    /// 统计记录数（支持过滤和搜索）
+    pub async fn count(
+        plugin_id: Uuid,
+        entity_name: &str,
+        tenant_id: Uuid,
+        db: &sea_orm::DatabaseConnection,
+        filter: Option<serde_json::Value>,
+        search: Option<String>,
+    ) -> AppResult<u64> {
+        let table_name = resolve_table_name(plugin_id, entity_name, tenant_id, db).await?;
+
+        // 获取 searchable 字段列表，构建搜索条件
+        let entity_fields = resolve_entity_fields(plugin_id, entity_name, tenant_id, db).await?;
+        let search_tuple = {
+            let searchable: Vec<&str> = entity_fields
+                .iter()
+                .filter(|f| f.searchable == Some(true))
+                .map(|f| f.name.as_str())
+                .collect();
+            match (searchable.is_empty(), &search) {
+                (false, Some(kw)) => Some((searchable.join(","), kw.clone())),
+                _ => None,
+            }
+        };
+
+        let (sql, values) = DynamicTableManager::build_filtered_count_sql(
+            &table_name,
+            tenant_id,
+            filter,
+            search_tuple,
+        )
+        .map_err(|e| AppError::Validation(e))?;
+
+        #[derive(FromQueryResult)]
+        struct CountResult {
+            count: i64,
+        }
+
+        let result = CountResult::find_by_statement(Statement::from_sql_and_values(
+            sea_orm::DatabaseBackend::Postgres,
+            sql,
+            values,
+        ))
+        .one(db)
+        .await?
+        .map(|r| r.count as u64)
+        .unwrap_or(0);
+
+        Ok(result)
+    }
+
+    /// 聚合查询 — 按字段分组计数
+    /// 返回 [(分组键, 计数), ...]
+    pub async fn aggregate(
+        plugin_id: Uuid,
+        entity_name: &str,
+        tenant_id: Uuid,
+        db: &sea_orm::DatabaseConnection,
+        group_by_field: &str,
+        filter: Option<serde_json::Value>,
+    ) -> AppResult<Vec<(String, i64)>> {
+        let table_name = resolve_table_name(plugin_id, entity_name, tenant_id, db).await?;
+
+        let (sql, values) = DynamicTableManager::build_aggregate_sql(
+            &table_name,
+            tenant_id,
+            group_by_field,
+            filter,
+        )
+        .map_err(|e| AppError::Validation(e))?;
+
+        #[derive(FromQueryResult)]
+        struct AggRow {
+            key: Option<String>,
+            count: i64,
+        }
+
+        let rows = AggRow::find_by_statement(Statement::from_sql_and_values(
+            sea_orm::DatabaseBackend::Postgres,
+            sql,
+            values,
+        ))
+        .all(db)
+        .await?;
+
+        let result = rows
+            .into_iter()
+            .map(|r| (r.key.unwrap_or_default(), r.count))
+            .collect();
+
+        Ok(result)
+    }
 }
 
 /// 从 plugin_entities 表解析 table_name（带租户隔离）
diff --git a/crates/erp-plugin/src/dynamic_table.rs b/crates/erp-plugin/src/dynamic_table.rs
index c7144ea..ee70990 100644
--- a/crates/erp-plugin/src/dynamic_table.rs
+++ b/crates/erp-plugin/src/dynamic_table.rs
@@ -287,6 +287,117 @@ impl DynamicTableManager {
         )
     }
 
+    /// 构建带过滤条件的 COUNT SQL
+    /// 复用 build_filtered_query_sql 的条件构建逻辑，但只做 COUNT
+    pub fn build_filtered_count_sql(
+        table_name: &str,
+        tenant_id: Uuid,
+        filter: Option<serde_json::Value>,
+        search: Option<(String, String)>, // (searchable_fields_csv, keyword)
+    ) -> Result<(String, Vec<Value>), String> {
+        let mut conditions = vec![
+            format!("\"tenant_id\" = ${}", 1),
+            "\"deleted_at\" IS NULL".to_string(),
+        ];
+        let mut param_idx = 2;
+        let mut values: Vec<Value> = vec![tenant_id.into()];
+
+        // 处理 filter（与 build_filtered_query_sql 保持一致）
+        if let Some(f) = filter {
+            if let Some(obj) = f.as_object() {
+                for (key, val) in obj {
+                    let clean_key = sanitize_identifier(key);
+                    if clean_key.is_empty() {
+                        return Err(format!("无效的过滤字段名: {}", key));
+                    }
+                    conditions.push(format!("\"data\"->>'{}' = ${}", clean_key, param_idx));
+                    values.push(Value::String(Some(Box::new(
+                        val.as_str().unwrap_or("").to_string(),
+                    ))));
+                    param_idx += 1;
+                }
+            }
+        }
+
+        // 处理 search（与 build_filtered_query_sql 保持一致）
+        if let Some((fields_csv, keyword)) = search {
+            let escaped = keyword.replace('%', "\\%").replace('_', "\\_");
+            let fields: Vec<&str> = fields_csv.split(',').collect();
+            let search_param_idx = param_idx;
+            let search_conditions: Vec<String> = fields
+                .iter()
+                .map(|f| {
+                    let clean = sanitize_identifier(f.trim());
+                    format!("\"data\"->>'{}' ILIKE ${}", clean, search_param_idx)
+                })
+                .collect();
+            conditions.push(format!("({})", search_conditions.join(" OR ")));
+            values.push(Value::String(Some(Box::new(format!("%{}%", escaped)))));
+        }
+
+        let sql = format!(
+            "SELECT COUNT(*) as count FROM \"{}\" WHERE {}",
+            table_name,
+            conditions.join(" AND "),
+        );
+
+        Ok((sql, values))
+    }
+
+    /// 构建聚合查询 SQL — 按 JSONB 字段分组计数
+    /// SELECT data->>'group_field' as key, COUNT(*) as count
+    /// FROM table WHERE tenant_id = $1 AND deleted_at IS NULL [AND filter...]
+    /// GROUP BY data->>'group_field' ORDER BY count DESC
+    pub fn build_aggregate_sql(
+        table_name: &str,
+        tenant_id: Uuid,
+        group_by_field: &str,
+        filter: Option<serde_json::Value>,
+    ) -> Result<(String, Vec<Value>), String> {
+        let clean_group = sanitize_identifier(group_by_field);
+        if clean_group.is_empty() {
+            return Err(format!("无效的分组字段名: {}", group_by_field));
+        }
+
+        let mut conditions = vec![
+            format!("\"tenant_id\" = ${}", 1),
+            "\"deleted_at\" IS NULL".to_string(),
+        ];
+        let mut param_idx = 2;
+        let mut values: Vec<Value> = vec![tenant_id.into()];
+
+        // 处理 filter
+        if let Some(f) = filter {
+            if let Some(obj) = f.as_object() {
+                for (key, val) in obj {
+                    let clean_key = sanitize_identifier(key);
+                    if clean_key.is_empty() {
+                        return Err(format!("无效的过滤字段名: {}", key));
+                    }
+                    conditions.push(format!("\"data\"->>'{}' = ${}", clean_key, param_idx));
+                    values.push(Value::String(Some(Box::new(
+                        val.as_str().unwrap_or("").to_string(),
+                    ))));
+                    param_idx += 1;
+                }
+            }
+        }
+
+        let sql = format!(
+            "SELECT \"data\"->>'{}' as key, COUNT(*) as count \
+             FROM \"{}\" \
+             WHERE {} \
+             GROUP BY \"data\"->>'{}' \
+             ORDER BY count DESC",
+            clean_group,
+            table_name,
+            conditions.join(" AND "),
+            clean_group,
+        );
+
+        Ok((sql, values))
+    }
+
     /// 构建带过滤条件的查询 SQL
     pub fn build_filtered_query_sql(
         table_name: &str,
@@ -492,4 +603,116 @@ mod tests {
             sql
         );
     }
+
+    // ===== build_filtered_count_sql 测试 =====
+
+    #[test]
+    fn test_build_filtered_count_sql_basic() {
+        let (sql, values) = DynamicTableManager::build_filtered_count_sql(
+            "plugin_test",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            None,
+            None,
+        )
+        .unwrap();
+        assert!(sql.contains("COUNT(*)"), "Expected COUNT(*), got: {}", sql);
+        assert!(sql.contains("tenant_id"), "Expected tenant_id filter");
+        assert!(sql.contains("deleted_at"), "Expected soft delete filter");
+        assert_eq!(values.len(), 1); // 仅 tenant_id
+    }
+
+    #[test]
+    fn test_build_filtered_count_sql_with_filter() {
+        let (sql, values) = DynamicTableManager::build_filtered_count_sql(
+            "plugin_test_customer",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            Some(serde_json::json!({"status": "active"})),
+            None,
+        )
+        .unwrap();
+        assert!(sql.contains("\"data\"->>'status' ="), "Expected filter, got: {}", sql);
+        assert_eq!(values.len(), 2); // tenant_id + filter_value
+    }
+
+    #[test]
+    fn test_build_filtered_count_sql_with_search() {
+        let (sql, values) = DynamicTableManager::build_filtered_count_sql(
+            "plugin_test",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            None,
+            Some(("name,code".to_string(), "搜索词".to_string())),
+        )
+        .unwrap();
+        assert!(sql.contains("ILIKE"), "Expected ILIKE, got: {}", sql);
+        assert_eq!(values.len(), 2); // tenant_id + search_param
+    }
+
+    #[test]
+    fn test_build_filtered_count_sql_no_limit_offset() {
+        // COUNT SQL 不应包含 LIMIT/OFFSET
+        let (sql, _) = DynamicTableManager::build_filtered_count_sql(
+            "plugin_test",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            None,
+            None,
+        )
+        .unwrap();
+        assert!(!sql.contains("LIMIT"), "COUNT SQL 不应包含 LIMIT");
+        assert!(!sql.contains("OFFSET"), "COUNT SQL 不应包含 OFFSET");
+    }
+
+    // ===== build_aggregate_sql 测试 =====
+
+    #[test]
+    fn test_build_aggregate_sql_basic() {
+        let (sql, values) = DynamicTableManager::build_aggregate_sql(
+            "plugin_test_customer",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            "status",
+            None,
+        )
+        .unwrap();
+        assert!(sql.contains("GROUP BY"), "Expected GROUP BY, got: {}", sql);
+        assert!(sql.contains("\"data\"->>'status'"), "Expected group field, got: {}", sql);
+        assert!(sql.contains("ORDER BY count DESC"), "Expected ORDER BY count DESC, got: {}", sql);
+        assert_eq!(values.len(), 1); // 仅 tenant_id
+    }
+
+    #[test]
+    fn test_build_aggregate_sql_with_filter() {
+        let (sql, values) = DynamicTableManager::build_aggregate_sql(
+            "plugin_test_customer",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            "region",
+            Some(serde_json::json!({"status": "active"})),
+        )
+        .unwrap();
+        assert!(sql.contains("\"data\"->>'region'"), "Expected group field, got: {}", sql);
+        assert!(sql.contains("\"data\"->>'status' ="), "Expected filter, got: {}", sql);
+        assert_eq!(values.len(), 2); // tenant_id + filter_value
+    }
+
+    #[test]
+    fn test_build_aggregate_sql_sanitizes_group_field() {
+        let result = DynamicTableManager::build_aggregate_sql(
+            "plugin_test",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            "evil'; DROP TABLE--",
+            None,
+        );
+        let (sql, _) = result.unwrap();
+        assert!(!sql.contains("DROP TABLE"), "SQL 不应包含注入: {}", sql);
+        assert!(sql.contains("evil___DROP_TABLE__"), "字段名应被清理: {}", sql);
+    }
+
+    #[test]
+    fn test_build_aggregate_sql_empty_field_rejected() {
+        let result = DynamicTableManager::build_aggregate_sql(
+            "plugin_test",
+            Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
+            "",
+            None,
+        );
+        assert!(result.is_err(), "空字段名应被拒绝");
+    }
 }
diff --git a/crates/erp-plugin/src/handler/data_handler.rs b/crates/erp-plugin/src/handler/data_handler.rs
index dd20bb1..fad865f 100644
--- a/crates/erp-plugin/src/handler/data_handler.rs
+++ b/crates/erp-plugin/src/handler/data_handler.rs
@@ -7,7 +7,10 @@ use erp_core::error::AppError;
 use erp_core::rbac::require_permission;
 use erp_core::types::{ApiResponse, PaginatedResponse, TenantContext};
 
-use crate::data_dto::{CreatePluginDataReq, PluginDataListParams, PluginDataResp, UpdatePluginDataReq};
+use crate::data_dto::{
+    AggregateItem, AggregateQueryParams, CountQueryParams, CreatePluginDataReq,
+    PluginDataListParams, PluginDataResp, UpdatePluginDataReq,
+};
 use crate::data_service::PluginDataService;
 use crate::state::PluginState;
 
@@ -228,3 +231,98 @@ where
 
     Ok(Json(ApiResponse::ok(())))
 }
+
+#[utoipa::path(
+    get,
+    path = "/api/v1/plugins/{plugin_id}/{entity}/count",
+    params(CountQueryParams),
+    responses(
+        (status = 200, description = "成功", body = ApiResponse<u64>),
+    ),
+    security(("bearer_auth" = [])),
+    tag = "插件数据"
+)]
+/// GET /api/v1/plugins/{plugin_id}/{entity}/count — 统计计数
+pub async fn count_plugin_data<S>(
+    State(state): State<PluginState>,
+    Extension(ctx): Extension<TenantContext>,
+    Path((plugin_id, entity)): Path<(Uuid, String)>,
+    Query(params): Query<CountQueryParams>,
+) -> Result<Json<ApiResponse<u64>>, AppError>
+where
+    PluginState: FromRef<S>,
+    S: Clone + Send + Sync + 'static,
+{
+    let fine_perm = compute_permission_code(&plugin_id.to_string(), &entity, "list");
+    if require_permission(&ctx, &fine_perm).is_err() {
+        require_permission(&ctx, "plugin.list")?;
+    }
+
+    // 解析 filter JSON
+    let filter: Option<serde_json::Value> = params
+        .filter
+        .as_ref()
+        .and_then(|f| serde_json::from_str(f).ok());
+
+    let total = PluginDataService::count(
+        plugin_id,
+        &entity,
+        ctx.tenant_id,
+        &state.db,
+        filter,
+        params.search,
+    )
+    .await?;
+
+    Ok(Json(ApiResponse::ok(total)))
+}
+
+#[utoipa::path(
+    get,
+    path = "/api/v1/plugins/{plugin_id}/{entity}/aggregate",
+    params(AggregateQueryParams),
+    responses(
+        (status = 200, description = "成功", body = ApiResponse<Vec<AggregateItem>>),
+    ),
+    security(("bearer_auth" = [])),
+    tag = "插件数据"
+)]
+/// GET /api/v1/plugins/{plugin_id}/{entity}/aggregate — 聚合查询
+pub async fn aggregate_plugin_data<S>(
+    State(state): State<PluginState>,
+    Extension(ctx): Extension<TenantContext>,
+    Path((plugin_id, entity)): Path<(Uuid, String)>,
+    Query(params): Query<AggregateQueryParams>,
+) -> Result<Json<ApiResponse<Vec<AggregateItem>>>, AppError>
+where
+    PluginState: FromRef<S>,
+    S: Clone + Send + Sync + 'static,
+{
+    let fine_perm = compute_permission_code(&plugin_id.to_string(), &entity, "list");
+    if require_permission(&ctx, &fine_perm).is_err() {
+        require_permission(&ctx, "plugin.list")?;
+    }
+
+    // 解析 filter JSON
+    let filter: Option<serde_json::Value> = params
+        .filter
+        .as_ref()
+        .and_then(|f| serde_json::from_str(f).ok());
+
+    let rows = PluginDataService::aggregate(
+        plugin_id,
+        &entity,
+        ctx.tenant_id,
+        &state.db,
+        &params.group_by,
+        filter,
+    )
+    .await?;
+
+    let items = rows
+        .into_iter()
+        .map(|(key, count)| AggregateItem { key, count })
+        .collect();
+
+    Ok(Json(ApiResponse::ok(items)))
+}
diff --git a/crates/erp-plugin/src/module.rs b/crates/erp-plugin/src/module.rs
index f9d5e2b..c0d9ebd 100644
--- a/crates/erp-plugin/src/module.rs
+++ b/crates/erp-plugin/src/module.rs
@@ -76,6 +76,15 @@ impl PluginModule {
                 get(crate::handler::data_handler::get_plugin_data::<S>)
                     .put(crate::handler::data_handler::update_plugin_data::<S>)
                     .delete(crate::handler::data_handler::delete_plugin_data::<S>),
+            )
+            // 数据统计路由
+            .route(
+                "/plugins/{plugin_id}/{entity}/count",
+                get(crate::handler::data_handler::count_plugin_data::<S>),
+            )
+            .route(
+                "/plugins/{plugin_id}/{entity}/aggregate",
+                get(crate::handler::data_handler::aggregate_plugin_data::<S>),
             );
 
         admin_routes.merge(data_routes)