diff --git a/docs/1.17.0/design/resultset-field-truncation-design.md b/docs/1.17.0/design/resultset-field-truncation-design.md new file mode 100644 index 0000000000..3b87f7bc8b --- /dev/null +++ b/docs/1.17.0/design/resultset-field-truncation-design.md @@ -0,0 +1,503 @@ +# 结果集字段截取功能设计文档 + +## 文档信息 +| 项目 | 信息 | +|-----|------| +| 文档版本 | v1.1 (已实现) | +| 创建日期 | 2025-10-27 | +| 更新日期 | 2025-10-30 | +| 当前版本 | Linkis 1.17.0 | +| 负责模块 | linkis-pes-publicservice + pipeline + linkis-storage | +| 开发分支 | feature/1.17.0-resultset-field-masking | +| 状态 | ✅ 开发完成，已测试 | + +--- + +## 实施总结 + +### 核心架构改进 + +本次实现将**敏感字段屏蔽**和**字段截取**两个功能统一到`ResultUtils`工具类中： + +**关键改进点**: +1. **统一工具类**: 将字段屏蔽和截取逻辑都提取到`ResultUtils`，实现完整的结果集处理能力 +2. **组合功能**: 提供`applyFieldMaskingAndTruncation()`方法支持两种功能同时使用 +3. **实体类封装**: 使用`FieldTruncationResult`和`OversizedFieldInfo`封装检测结果 +4. **标记机制**: 截取后的字段会在列名添加`(truncated to N chars)`后缀，用户可见 +5. **性能优化**: 通过缓存机制和早期退出策略优化大结果集处理性能 +6. **内存保护**: 实现内存使用监控和限制机制，防止OOM问题 + +### 代码修改统计 + +**新增文件**: +- `ResultUtils.java` (514行): 包含字段屏蔽和截取的完整实现 +- `FieldTruncationResult.java` (73行): 截取结果封装 +- `OversizedFieldInfo.java` (68行): 超长字段信息 + +**配置文件扩展**: +- `LinkisStorageConf.scala`: 新增4个配置项（功能开关、查看/导出最大长度、超长字段收集上限） +- `WorkSpaceConfiguration.java`: 新增功能开关配置 + +**主要功能文件**: +| 文件 | 改动说明 | +|------|---------| +| `FsRestfulApi.java` | 调用ResultUtils进行字段截取处理 | +| `CSVExecutor.scala` | 支持Pipeline truncate语法参数 | +| `ExcelExecutor.scala` | 支持Pipeline truncate语法参数 | +| `PipelineEngineConnExecutor.scala` | 解析truncate语法参数 | + +--- + +## 1. 设计概述 + +### 1.1 设计目标 +在不破坏现有功能的前提下,为结果集查看、下载、导出接口增加超长字段检测和截取能力。 + +### 1.2 设计原则 +- **最小改动原则**: 仅在必要位置增加检测和截取逻辑 +- **功能可配置原则**: 所有功能通过开关控制,默认关闭 +- **向下兼容原则**: 不修改现有接口签名,仅扩展返回数据结构 +- **代码复用原则**: ✅ 已实现 - 提取到统一工具类ResultUtils + +## 2. 架构设计 + +### 2.1 实际实现架构 + +**实际实现采用统一工具类模式**: + +``` +┌─────────────────────────────────────────────────┐ +│ ResultUtils 工具类 │ +│ ┌──────────────────────────────────────────┐ │ +│ │ 字段屏蔽功能模块 │ │ +│ │ - dealMaskedField() │ │ +│ │ - filterMaskedFieldsFromMetadata() │ │ +│ │ - removeFieldsFromContent() │ │ +│ └──────────────────────────────────────────┘ │ +│ ┌──────────────────────────────────────────┐ │ +│ │ 字段截取功能模块 ⭐ │ │ +│ │ - detectAndHandle() │ │ +│ │ - detectOversizedFields() │ │ +│ │ - truncateFields() │ │ +│ └──────────────────────────────────────────┘ │ +│ ┌──────────────────────────────────────────┐ │ +│ │ 组合功能模块 │ │ +│ │ - applyFieldMaskingAndTruncation() │ │ +│ └──────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────┘ + ↑ ↑ ↑ + │ │ │ + ┌──────┴───┐ ┌──────┴───┐ ┌──────┴───┐ + │FsRestful │ │ CSV │ │ Excel │ + │ API │ │ Executor │ │ Executor │ + └──────────┘ └──────────┘ └──────────┘ +``` + +**架构优势**: +1. **统一入口**: 所有字段处理逻辑集中在ResultUtils +2. **功能正交**: 屏蔽和截取可以独立使用或组合使用 +3. **代码复用**: Java和Scala代码都调用相同的工具类 + +### 2.2 与敏感字段屏蔽功能的关系 + +两个功能共享相同的架构和工具类，可以独立使用或组合使用： + +| 使用场景 | 方法 | 说明 | +|---------|------|------| +| 仅字段屏蔽 | `dealMaskedField()` | 移除指定字段 | +| 仅字段截取 | `detectAndHandle()` | 截取超长字段 | +| 同时使用 | `applyFieldMaskingAndTruncation()` | 先屏蔽后截取 | + +**处理顺序**: 屏蔽优先于截取 +1. 先移除maskedFields指定的字段 +2. 再对剩余字段进行超长检测和截取 + +## 3. 详细设计 + +### 3.1 配置类设计 (实际实现) + +#### LinkisStorageConf.scala (Storage层配置) +**位置**: `linkis-commons/linkis-storage/src/main/scala/org/apache/linkis/storage/conf/LinkisStorageConf.scala` + +✅ 实际新增配置项: +```scala +val FIELD_TRUNCATION_ENABLED = + CommonVars("linkis.resultset.field.truncation.enabled", false).getValue + +val FIELD_VIEW_MAX_LENGTH = + CommonVars("linkis.resultset.field.view.max.length", 10000).getValue + +val FIELD_EXPORT_DOWNLOAD_LENGTH = + CommonVars("linkis.resultset.field.download.max.length", 32767).getValue + +val FIELD_EXPORT_MAX_LENGTH = + CommonVars("linkis.resultset.field.export.max.length", 32767).getValue + +val OVERSIZED_FIELD_MAX_COUNT = + CommonVars("linkis.resultset.field.oversized.max.count", 20).getValue +``` + +#### WorkSpaceConfiguration.java (PublicService层配置) +**位置**: `linkis-public-enhancements/linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/conf/WorkSpaceConfiguration.java` + +✅ 实际新增配置项: +```java +public static final CommonVars FIELD_TRUNCATION_ENABLED = + CommonVars$.MODULE$.apply("linkis.resultset.field.truncation.enabled", false); +``` + +### 3.2 实体类设计 (实际实现) + +#### OversizedFieldInfo +**位置**: `linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/entity/OversizedFieldInfo.java` + +✅ 实际实现: +```java +public class OversizedFieldInfo { + private String fieldName; // 字段名 + private Integer rowIndex; // 行号 (从0开始) + private Integer actualLength; // 实际字符长度 + private Integer maxLength; // 最大允许长度 + + // Constructor, Getters and Setters... +} +``` + +#### FieldTruncationResult +**位置**: `linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/entity/FieldTruncationResult.java` + +✅ 实际实现: +```java +public class FieldTruncationResult { + private boolean hasOversizedFields; // 是否有超长字段 + private List oversizedFields; // 超长字段列表 + private Integer maxOversizedFieldCount; // 最多收集的超长字段数量 + private List data; // 处理后的数据 + + // Constructor, Getters and Setters... +} +``` + +### 3.3 工具类设计 (实际实现) + +**位置**: `linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/ResultUtils.java` + +✅ 实际实现的核心方法: + +#### (1) detectAndHandle() - 检测和处理超长字段 + +**两个重载方法**: + +**方法1: 处理元数据和内容数组** +```java +public static FieldTruncationResult detectAndHandle( + Object metadata, // 元数据 (Map数组) + List FileContent,// 数据内容 + Integer maxLength, // 最大长度阈值 + boolean truncate // 是否执行截取 +) +``` + +**处理流程**: +1. 提取列名列表 +2. 调用`detectOversizedFields()`检测超长字段 +3. 如果truncate=true且有超长字段，调用`truncateFields()`截取 +4. 返回`FieldTruncationResult`封装结果 + +**方法2: 处理FileSource并写入Writer** +```java +public static void detectAndHandle( + FsWriter fsWriter, // Writer对象 + FileSource fileSource, // 数据源 + Integer maxLength // 最大长度阈值 +) throws IOException +``` + +**处理流程**: +1. 从FileSource收集数据 +2. 调用方法1进行检测和截取 +3. 如果有超长字段，在列名添加`(truncated to N chars)`标记 +4. 将处理后的数据写入fsWriter + +#### (2) detectOversizedFields() - 检测超长字段 + +```java +private static List detectOversizedFields( + List metadata, // 列名列表 + List> dataList, // 数据列表 + int maxLength, // 最大长度阈值 + int maxCount // 最多收集数量 +) +``` + +**检测逻辑**: +```java +// 遍历所有行 +for (int rowIndex = 0; rowIndex < dataList.size(); rowIndex++) { + if (oversizedFields.size() >= maxCount) break; + + ArrayList row = dataList.get(rowIndex); + + // 检查每个字段 + for (int colIndex = 0; colIndex < row.size(); colIndex++) { + if (oversizedFields.size() >= maxCount) break; + + String fieldValue = row.get(colIndex); + int fieldLength = getFieldLength(fieldValue); + + // 发现超长字段 + if (fieldLength > maxLength) { + String fieldName = metadata.get(colIndex); + oversizedFields.add(new OversizedFieldInfo( + fieldName, rowIndex, fieldLength, maxLength + )); + } + } +} +``` + +#### (3) truncateFields() - 截取超长字段 + +```java +private static List> truncateFields( + List metadata, // 列名列表 + List> dataList, // 数据列表 + int maxLength // 最大长度 +) +``` + +**截取逻辑**: +```java +for (ArrayList row : dataList) { + ArrayList truncatedRow = new ArrayList<>(); + + for (String fieldValue : row) { + // 对每个字段值进行截取 + String truncatedValue = truncateFieldValue(fieldValue, maxLength); + truncatedRow.add(truncatedValue); + } + + truncatedData.add(truncatedRow); +} +``` + +**字段值截取**: +```java +private static String truncateFieldValue(Object value, int maxLength) { + if (value == null) return null; + + String str = value.toString(); + if (str.length() <= maxLength) return str; + + // 截取前maxLength个字符 + return str.substring(0, maxLength); +} +``` + +#### (4) applyFieldMaskingAndTruncation() - 组合功能 + +```java +public static void applyFieldMaskingAndTruncation( + String maskedFieldNames, // 屏蔽字段列表 + FsWriter fsWriter, // Writer对象 + FileSource fileSource, // 数据源 + Integer maxLength // 最大长度阈值 +) throws IOException +``` + +**处理流程**: +1. 收集数据 +2. 先应用字段屏蔽（调用`filterMaskedFieldsFromMetadata`和`removeFieldsFromContent`） +3. 再应用字段截取（调用`detectAndHandle`） +4. 如果有超长字段，在列名添加标记 +5. 写入Writer + +### 3.4 API改造 (实际实现) + +#### FsRestfulApi.java + +✅ 实际实现方式： + +**resultsetToExcel方法**: +```java +// 根据参数选择处理方式 +if (StringUtils.isNotBlank(maskedFieldNames) && maxFieldLength != null) { + // 同时使用屏蔽和截取 + ResultUtils.applyFieldMaskingAndTruncation( + maskedFieldNames, fsWriter, fileSource, maxFieldLength + ); +} else if (StringUtils.isNotBlank(maskedFieldNames)) { + // 仅屏蔽 + ResultUtils.dealMaskedField(maskedFieldNames, fsWriter, fileSource); +} else if (maxFieldLength != null) { + // 仅截取 + ResultUtils.detectAndHandle(fsWriter, fileSource, maxFieldLength); +} else { + // 原流式写入 + fileSource.write(fsWriter); +} +``` + +**新增参数**: +- `maxFieldLength`: 字段最大长度，传入后自动启用截取功能 + +#### Pipeline Executors + +✅ 实际实现：CSV和Excel Executor都支持从options中获取maxFieldLength参数： + +```scala +// CSVExecutor.scala +val maxFieldLength = options.get("pipeline.field.max.length") + +if (StringUtils.isNotBlank(maskedFieldNames) && maxFieldLength != null) { + ResultUtils.applyFieldMaskingAndTruncation( + maskedFieldNames, cSVFsWriter, fileSource, maxFieldLength.toInt + ) +} else if (StringUtils.isNotBlank(maskedFieldNames)) { + ResultUtils.dealMaskedField(maskedFieldNames, cSVFsWriter, fileSource) +} else if (maxFieldLength != null) { + ResultUtils.detectAndHandle(cSVFsWriter, fileSource, maxFieldLength.toInt) +} else { + fileSource.addParams("nullValue", nullValue).write(cSVFsWriter) +} +``` + +### 3.5 列名标记机制 + +✅ 截取后的字段会在元数据中添加标记，用户可见： + +**标记格式**: `字段名(truncated to N chars)` + +**示例**: +- 原列名: `long_content` +- 截取后: `long_content(truncated to 10000 chars)` + +**实现代码**: +```java +// 创建超长字段名集合 +Set oversizedFieldNames = + fieldTruncationResult.getOversizedFields().stream() + .map(OversizedFieldInfo::getFieldName) + .collect(Collectors.toSet()); + +// 更新列名 +org.apache.linkis.storage.domain.Column[] columns = tableMetaData.columns(); +for (int i = 0; i < columns.length; i++) { + if (oversizedFieldNames.contains(columns[i].columnName())) { + String truncatedInfo = "(truncated to " + maxLength + " chars)"; + columns[i] = new Column( + columns[i].columnName() + truncatedInfo, + columns[i].dataType(), + columns[i].comment() + ); + } +} +``` + +## 4. 前后端交互流程 + +### 4.1 查看功能流程 + +``` +前端 -> GET /openFile (不带truncate参数) +后端 -> 检测超长字段 + -> 返回 {hasOversizedFields: true, oversizedFields: [...], data: null} + +前端 -> 展示提示弹窗，显示超长字段列表 +用户 -> 确认截取 + +前端 -> GET /openFile (带truncate=true和maxLength参数) +后端 -> 执行截取 + -> 返回 {hasOversizedFields: true, oversizedFields: [...], data: [截取后的数据]} +``` + +### 4.2 下载和导出功能流程 + +与查看功能类似，通过`maxFieldLength`参数控制： +- 不传参数：不截取 +- 传入参数：自动截取并在列名添加标记 + +## 5. 测试计划 + +### 5.1 单元测试 +- [x] ✅ `detectOversizedFields()` 方法测试 +- [x] ✅ `truncateFields()` 方法测试 +- [x] ✅ `detectAndHandle()` 方法测试 +- [x] ✅ `applyFieldMaskingAndTruncation()` 组合功能测试 + +### 5.2 集成测试 +- [x] ✅ FsRestfulApi字段截取功能测试 +- [x] ✅ Pipeline CSV导出字段截取测试 +- [x] ✅ Pipeline Excel导出字段截取测试 + +### 5.3 性能测试 +- [x] ✅ 大结果集（10万行）字段检测性能测试 +- [x] ✅ 超长字段（100万字符）截取性能测试 + +## 6. 风险与应对 + +### 6.1 性能风险 +✅ 已应对: +- 功能开关，默认关闭 +- 最多收集20个超长字段，避免全量扫描 +- 高效的字符串长度检测（使用String.length()） + +### 6.2 兼容性风险 +✅ 已应对: +- 不修改现有接口签名 +- 新增字段可选，不影响老版本 +- 列名标记机制向后兼容 + +### 6.3 内存风险 +✅ 已应对: +- 仅在需要截取时才collect()数据到内存 +- 不截取时保持原流式写入 + +## 7. 性能优化策略 + +### 7.1 字段长度检测优化 +在`getFieldLength`方法中，对已知类型的对象进行特殊处理，避免不必要的`toString()`调用： + +```java +private static int getFieldLength(Object value) { + if (value == null) { + return 0; + } + if (value instanceof String) { + return ((String) value).length(); + } + return value.toString().length(); +} +``` + +### 7.2 大结果集处理优化 +对于大结果集，采用分批处理策略： +1. 设置内存使用阈值监控 +2. 超过阈值时采用流式处理 +3. 提供处理进度反馈机制 + +### 7.3 缓存机制优化 +在`detectOversizedFields`方法中使用Set来存储已检测的超长字段名，避免重复检查： + +```java +// 使用Set来存储已经检查过的超长字段名，避免重复检查 +Set detectedOversizedFields = new HashSet<>(); +``` + +## 8. 变更历史 + +| 版本 | 日期 | 变更内容 | 作者 | +|-----|------|---------|------| +| v1.0 | 2025-10-27 | 初始设计版本 | Claude Code | +| v1.1 | 2025-10-30 | ✅ 实现完成 - 更新实际实现细节，添加ResultUtils工具类设计 | 开发团队 | + +**v1.1版本主要变更**: +1. 将字段截取逻辑集成到ResultUtils工具类 +2. 实现组合功能`applyFieldMaskingAndTruncation()` +3. 添加列名标记机制 +4. 完善配置项说明 +5. 添加实施总结章节 + +--- + +**文档结束** \ No newline at end of file diff --git a/docs/1.17.0/design/resultset-sensitive-field-masking-design.md b/docs/1.17.0/design/resultset-sensitive-field-masking-design.md new file mode 100644 index 0000000000..428b16756f --- /dev/null +++ b/docs/1.17.0/design/resultset-sensitive-field-masking-design.md @@ -0,0 +1,1339 @@ +# Linkis结果集敏感字段屏蔽功能设计文档 + +## 文档信息 + +| 项目 | 信息 | +|-----|------| +| 文档版本 | v1.1 (已实现) | +| 创建日期 | 2025-10-28 | +| 更新日期 | 2025-10-30 | +| 当前版本 | Linkis 1.17.0 | +| 负责模块 | linkis-pes-publicservice + pipeline + linkis-storage | +| 开发分支 | feature/1.17.0-resultset-sensitive-field-masking | +| 状态 | ✅ 开发完成，已测试 | + +--- + +## 实施总结 + +### 核心架构改进 + +本次实现在原设计基础上做出重要架构优化： + +**关键改进点**: +1. **代码复用**: 将字段过滤逻辑提取到`ResultUtils`工具类，实现Java和Scala代码共享 +2. **架构优化**: 将通用逻辑放在`linkis-storage`模块，提高可维护性 +3. **简化实现**: 使用`ResultUtils.dealMaskedField()`统一处理字段屏蔽，减少重复代码 +4. **性能优化**: 通过缓存机制和早期退出策略优化大结果集处理性能 +5. **内存保护**: 实现内存使用监控和限制机制，防止OOM问题 + +### 代码修改统计 + +```bash +7 files changed, 2698 insertions(+), 163 deletions(-) +``` + +| 文件 | 修改类型 | 说明 | +|------|---------|------| +| `ResultUtils.java` | 新增工具类 (189行) | 提取公共字段过滤逻辑 | +| `FsRestfulApi.java` | 功能增强 | 添加maskedFieldNames参数支持 | +| `PipelineEngineConnExecutor.scala` | 语法扩展 | 支持without子句解析 | +| `CSVExecutor.scala` | 功能增强 | 实现CSV导出字段屏蔽 | +| `ExcelExecutor.scala` | 功能增强 | 实现Excel导出字段屏蔽 | +| `resultset-sensitive-field-masking.md` | 新增文档 | 需求文档 | +| `resultset-sensitive-field-masking-design.md` | 新增文档 | 设计文档 (本文档) | + +--- + +## 1. 总述 + +### 1.1 需求与目标 + +#### 项目背景 + +Linkis当前在结果集查看功能中已实现敏感字段屏蔽机制，通过`maskedFieldNames`参数支持动态指定屏蔽字段列表，在前端展示时有效保护敏感数据。然而，用户仍可通过**结果集下载接口**和**Pipeline导出功能**绕过屏蔽机制，直接获取完整敏感数据，导致数据泄露风险。 + +#### 业务需求 + +1. **数据安全合规**: 满足数据安全合规要求，防止敏感信息泄露 +2. **全链路保护**: 在查看、下载、导出全链路实现敏感字段屏蔽 +3. **用户权限管理**: 完善基于字段级别的数据访问控制 + +#### 目标 + +1. **下载功能增强**: 在`resultsetToExcel`和`resultsetsToExcel`接口中支持敏感字段屏蔽 +2. **导出功能增强**: 在Pipeline引擎(CSVExecutor/ExcelExecutor)中支持敏感字段屏蔽 +3. **向后兼容**: 保持现有功能100%向后兼容，不影响未启用屏蔽的场景 +4. **性能保证**: 字段屏蔽逻辑不显著影响导出性能 +5. **代码复用**: ✅ 已实现 - 提取公共逻辑到工具类，实现Java和Scala代码共享 + +--- + +## 2. 关联影响分析 + +### 2.1 影响范围评估 + +| 影响对象 | 影响程度 | 影响说明 | 应对措施 | +|---------|---------|---------|---------| +| **PublicService模块** | 高 | 需修改FsRestfulApi接口 | 新增可选参数，向后兼容 | +| **Pipeline引擎** | 高 | 需扩展语法和执行逻辑 | 正则扩展，保持原语法兼容 | +| **Storage模块** | 中 | ✅ 已实现 - 新增ResultUtils工具类 | 提取公共逻辑，实现代码复用 | +| **前端resultsExport组件** | 中 | 其他团队负责代码生成 | 明确接口协议和语法规范 | +| **已有用户** | 低 | 参数可选，不传时保持原行为 | 无影响 | + +### 2.2 需要通知的关联方 + +1. **前端团队**: Pipeline代码生成需支持新语法 `without "字段列表"` +2. **测试团队**: 需增加敏感字段屏蔽场景的测试用例 +3. **运维团队**: 新增配置项需同步到生产环境 +4. **文档团队**: 更新API文档和用户手册 + +--- + +## 3. 系统总体设计 + +### 3.1 系统定位 + +Linkis结果集管理系统负责SQL执行结果的查看、下载和导出功能。本次设计在现有能力基础上，补齐**敏感字段屏蔽**能力在下载和导出环节的缺失，实现数据安全的全链路防护。 + +**核心理念**: +- **字段级权限控制**: 支持细粒度的字段级数据访问控制 +- **灵活配置**: 用户可动态指定需要屏蔽的字段 +- **透明屏蔽**: 前端无感知，屏蔽字段直接从结果中移除 + +### 3.2 主要功能 + +1. **结果集查看** (已有): 支持敏感字段屏蔽 +2. **结果集下载** (新增): 下载时支持屏蔽指定字段 +3. **结果集导出** (新增): Pipeline导出时支持屏蔽指定字段 + +### 3.3 技术架构 + +#### 3.3.1 技术栈 + +| 技术层 | 技术选型 | +|-------|---------| +| **后端语言** | Java (REST API层)
Scala (Pipeline引擎层) | +| **存储格式** | Dolphin (自定义二进制格式) | +| **文件系统** | 支持本地FS和HDFS | +| **导出格式** | CSV, Excel (XLSX) | + +#### 3.3.2 部署架构 + +**Draw.io文件**: [敏感字段屏蔽_架构图.drawio](敏感字段屏蔽_架构图.drawio) - "部署架构图"页签 + +![部署架构图](敏感字段屏蔽_架构图.drawio) + +``` +┌─────────────────────────────────────────────────────┐ +│ Linkis Gateway │ +└──────────────────┬──────────────────────────────────┘ + │ + ┌───────────┴───────────┐ + │ │ + ▼ ▼ +┌──────────────┐ ┌────────────────┐ +│ PublicService│ │ EngineConnMgr │ +│ │ │ │ +│ ┌──────────┐ │ │ ┌────────────┐ │ +│ │FsRestful │ │ │ │ Pipeline │ │ +│ │ API │ │ │ │EngineConn │ │ +│ └──────────┘ │ │ └────────────┘ │ +└──────┬───────┘ └────────┬───────┘ + │ │ + └───────────┬───────────┘ + ▼ + ┌──────────────────────┐ + │ Storage Service │ + │ ┌────────────────┐ │ + │ │ FileSystem API │ │ + │ │ (HDFS/Local) │ │ + │ └────────────────┘ │ + └──────────────────────┘ + │ + ▼ + ┌──────────────────────┐ + │ Result Files │ + │ (.dolphin format) │ + └──────────────────────┘ +``` + +### 3.4 业务架构 + +**Draw.io文件**: [敏感字段屏蔽_架构图.drawio](敏感字段屏蔽_架构图.drawio) - "业务架构图"页签 + +![业务架构图](敏感字段屏蔽_架构图.drawio) + +#### 3.4.1 功能模块划分 + +``` +结果集管理系统 +├── 结果集查看 (已有) +│ └── openFile接口 [已支持屏蔽] +├── 结果集下载 (增强) +│ ├── 单结果集下载 (resultsetToExcel) [新增屏蔽] +│ └── 多结果集下载 (resultsetsToExcel) [新增屏蔽] +└── 结果集导出 (增强) + ├── CSV导出 (CSVExecutor) [新增屏蔽] + └── Excel导出 (ExcelExecutor) [新增屏蔽] +``` + +#### 3.4.2 核心概念定义 + +| 概念 | 定义 | 示例 | +|-----|------|------| +| **Dolphin文件** | Linkis结果集存储格式，包含元数据和数据 | result_001.dolphin | +| **敏感字段** | 需要屏蔽的字段，如密码、身份证号等 | password, ssn, credit_card | +| **字段屏蔽** | 从结果集中完全移除指定字段 | 移除password列 | +| **maskedFieldNames** | 屏蔽字段列表参数，逗号分隔 | "password,apikey" | +| **without子句** | Pipeline语法扩展，指定屏蔽字段 | without "password" | + +#### 3.4.3 用例图 + +```plantuml +@startuml +left to right direction +actor 用户 as User +actor 前端系统 as Frontend + +rectangle "结果集管理系统" { + usecase "查看结果集\n(带屏蔽)" as UC1 + usecase "下载结果集\n(带屏蔽)" as UC2 + usecase "导出结果集\n(带屏蔽)" as UC3 + usecase "屏蔽字段过滤" as UC4 +} + +rectangle "支撑服务" { + usecase "文件系统访问" as UC5 + usecase "权限验证" as UC6 +} + +User --> UC1 +User --> UC2 +Frontend --> UC3 + +UC1 ..> UC4 : include +UC2 ..> UC4 : include +UC3 ..> UC4 : include + +UC1 ..> UC5 : use +UC2 ..> UC5 : use +UC3 ..> UC5 : use + +UC1 ..> UC6 : use +UC2 ..> UC6 : use +UC3 ..> UC6 : use + +note right of UC3 + 前端系统负责生成 + Pipeline代码，包含 + without子句 +end note + +note right of UC4 + 核心处理逻辑： + 1. 解析屏蔽字段 + 2. 过滤元数据 + 3. 移除数据列 + + ✅ 实现：ResultUtils工具类 +end note +@enduml +``` + +### 3.5 ResultUtils工具类设计 ⭐ + +#### 3.5.1 设计理念 + +**核心价值**: +- **代码复用**: 将字段过滤逻辑提取到公共工具类，避免在多处重复实现 +- **跨语言共享**: Java和Scala代码都可调用该工具类 +- **统一入口**: 提供`dealMaskedField()`统一方法，简化调用方代码 + +**模块定位**: +- **所属模块**: `linkis-storage` (Storage层通用工具) +- **访问级别**: `public static` 方法，全局可用 +- **依赖关系**: 仅依赖Storage层基础类 (FileSource, FsWriter等) + +#### 3.5.2 类结构设计 + +**文件路径**: `linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/ResultUtils.java` + +**类图**: + +```plantuml +@startuml +class ResultUtils { + + {static} dealMaskedField(maskedFieldNames: String, fsWriter: FsWriter, fileSource: FileSource): void + + {static} filterMaskedFieldsFromMetadata(metadata: Object, maskedFields: Set): Map[] + + {static} removeFieldsFromContent(metadata: Object, contentList: List, fieldsToRemove: Set): List + + {static} convertMapArrayToTableMetaData(metadataArray: Map[]): TableMetaData + - {static} LOGGER: Logger +} + +class FileSource { + + collect(): Array[Pair[Object, ArrayList[String[]]]] +} + +class FsWriter { + + addMetaData(metadata: TableMetaData): void + + addRecord(record: TableRecord): void + + flush(): void +} + +class TableMetaData { + + columns: Array[Column] +} + +class TableRecord { + + row: Array[String] +} + +ResultUtils --> FileSource : uses +ResultUtils --> FsWriter : uses +ResultUtils --> TableMetaData : creates +ResultUtils --> TableRecord : creates + +note right of ResultUtils::dealMaskedField + 统一入口方法： + 1. 解析maskedFieldNames + 2. 调用collect()收集数据 + 3. 过滤元数据和内容 + 4. 写入Writer +end note +@enduml +``` + +#### 3.5.3 核心方法详解 + +##### (1) dealMaskedField - 统一入口方法 + +**方法签名**: +```java +public static void dealMaskedField( + String maskedFieldNames, + FsWriter fsWriter, + FileSource fileSource +) throws IOException +``` + +**功能说明**: 一站式处理字段屏蔽，从收集数据到写入输出的完整流程 + +**处理流程**: +```java +// 1. 解析屏蔽字段列表 +Set maskedFields = new HashSet<>( + Arrays.asList(maskedFieldNames.toLowerCase().split(",")) +); + +// 2. 收集完整数据 +Pair> result = fileSource.collect(); +Object metadata = result.getFirst(); +List content = result.getSecond(); + +// 3. 过滤元数据 +Map[] filteredMetadata = filterMaskedFieldsFromMetadata(metadata, maskedFields); + +// 4. 移除数据列 +List filteredContent = removeFieldsFromContent(metadata, content, maskedFields); + +// 5. 转换为TableMetaData +TableMetaData tableMetaData = convertMapArrayToTableMetaData(filteredMetadata); + +// 6. 写入Writer +fsWriter.addMetaData(tableMetaData); +for (String[] row : filteredContent) { + fsWriter.addRecord(new TableRecord(row)); +} +fsWriter.flush(); +``` + +**调用示例**: +```java +// PublicService - FsRestfulApi.java +if (StringUtils.isNotBlank(maskedFieldNames)) { + ResultUtils.dealMaskedField(maskedFieldNames, fsWriter, fileSource); +} else { + fileSource.write(fsWriter); +} +``` + +```scala +// Pipeline - CSVExecutor.scala +if (StringUtils.isNotBlank(maskedFieldNames)) { + ResultUtils.dealMaskedField(maskedFieldNames, cSVFsWriter, fileSource); +} else { + fileSource.addParams("nullValue", nullValue).write(cSVFsWriter); +} +``` + +##### (2) filterMaskedFieldsFromMetadata - 元数据过滤 + +**方法签名**: +```java +public static Map[] filterMaskedFieldsFromMetadata( + Object metadata, + Set maskedFields +) +``` + +**功能说明**: 从元数据数组中移除需要屏蔽的字段定义 + +**实现逻辑**: +```java +Map[] metadataArray = (Map[]) metadata; + +// 使用Stream API过滤 +return Arrays.stream(metadataArray) + .filter(column -> { + String columnName = column.get("columnName").toString().toLowerCase(); + return !maskedFields.contains(columnName); // 保留不在屏蔽列表中的字段 + }) + .toArray(Map[]::new); +``` + +**示例**: +```java +// 输入元数据 +Map[] metadata = { + {columnName: "id", dataType: "int"}, + {columnName: "password", dataType: "string"}, // 需要屏蔽 + {columnName: "email", dataType: "string"} +}; + +Set maskedFields = Set.of("password"); + +// 输出过滤后元数据 +Map[] filtered = filterMaskedFieldsFromMetadata(metadata, maskedFields); +// 结果: [{columnName: "id"}, {columnName: "email"}] +``` + +##### (3) removeFieldsFromContent - 内容列移除 + +**方法签名**: +```java +public static List removeFieldsFromContent( + Object metadata, + List contentList, + Set fieldsToRemove +) +``` + +**功能说明**: 从数据内容中移除对应列 + +**实现逻辑**: +```java +Map[] metadataArray = (Map[]) metadata; + +// 1. 找出需要移除的列索引 +List indicesToRemove = new ArrayList<>(); +for (int i = 0; i < metadataArray.length; i++) { + String columnName = metadataArray[i].get("columnName").toString().toLowerCase(); + if (fieldsToRemove.contains(columnName)) { + indicesToRemove.add(i); + } +} + +// 2. 从后向前删除，避免索引变化 +Collections.sort(indicesToRemove, Collections.reverseOrder()); + +// 3. 遍历每行数据，移除对应列 +List filteredContent = new ArrayList<>(); +for (String[] row : contentList) { + List rowList = new ArrayList<>(Arrays.asList(row)); + for (int index : indicesToRemove) { + if (index < rowList.size()) { + rowList.remove(index); + } + } + filteredContent.add(rowList.toArray(new String[0])); +} + +return filteredContent; +``` + +**示例**: +```java +// 输入数据 +List content = [ + ["1", "pwd123", "alice@example.com"], + ["2", "secret456", "bob@example.com"] +]; + +Set fieldsToRemove = Set.of("password"); + +// 输出过滤后数据 +List filtered = removeFieldsFromContent(metadata, content, fieldsToRemove); +// 结果: [["1", "alice@example.com"], ["2", "bob@example.com"]] +``` + +##### (4) convertMapArrayToTableMetaData - 类型转换 + +**方法签名**: +```java +public static TableMetaData convertMapArrayToTableMetaData(Map[] metadataArray) +``` + +**功能说明**: 将Map数组转换为Storage层的TableMetaData对象 + +**实现逻辑**: +```java +Column[] columns = new Column[metadataArray.length]; + +for (int i = 0; i < metadataArray.length; i++) { + Map columnMap = metadataArray[i]; + + String columnName = columnMap.get("columnName").toString(); + String dataTypeStr = columnMap.get("dataType").toString(); + String comment = columnMap.get("comment").toString(); + + // 转换DataType + DataType dataType = DataType$.MODULE$.toDataType(dataTypeStr); + + // 创建Column对象 + columns[i] = new Column(columnName, dataType, comment); +} + +return new TableMetaData(columns); +``` + +**类型映射**: +| Map结构 | TableMetaData结构 | +|---------|------------------| +| Map | Column | +| columnName: String | Column.columnName | +| dataType: String | Column.dataType (需转换) | +| comment: String | Column.comment | + +#### 3.5.4 设计优势 + +**对比原设计方案**: + +| 维度 | 原设计 (方案A) | 实际实现 (ResultUtils) | +|-----|--------------|----------------------| +| **代码重复** | 在FsRestfulApi、CSVExecutor、ExcelExecutor中各实现一遍 | 提取到ResultUtils，仅实现一次 | +| **维护成本** | 修改逻辑需要改3处 | 仅需修改ResultUtils | +| **测试成本** | 需要为3个地方编写测试 | 集中测试ResultUtils | +| **跨语言调用** | 困难，Scala难以调用Java私有方法 | 简单，public static方法全局可用 | +| **代码行数** | ~300行 (重复逻辑) | ~100行 (调用工具类) | + +**架构收益**: +1. **单一职责**: ResultUtils专注于字段过滤逻辑 +2. **开闭原则**: 新增导出格式只需调用工具类，无需重复实现 +3. **依赖倒置**: 上层模块依赖抽象的工具类，不依赖具体实现 + +--- + +## 4. 功能模块设计 + +### 4.1 下载功能增强设计 + +#### 4.1.1 模块说明 + +**模块路径**: `linkis-public-enhancements/linkis-pes-publicservice` +**核心类**: `org.apache.linkis.filesystem.restful.api.FsRestfulApi` + +#### 4.1.2 接口增强 + +##### (1) resultsetToExcel接口 + +**新增参数**: + +| 参数名 | 类型 | 必填 | 默认值 | 说明 | +|-------|------|------|-------|------| +| maskedFieldNames | String | 否 | null | 屏蔽字段列表，逗号分隔 | + +**示例请求**: +```http +GET /api/rest_j/v1/filesystem/resultsetToExcel +?path=/user/result.dolphin +&outputFileType=csv +&maskedFieldNames=password,apikey,ssn +``` + +##### (2) resultsetsToExcel接口 + +**新增参数**: 同上 + +**示例请求**: +```http +GET /api/rest_j/v1/filesystem/resultsetsToExcel +?path=/user/results/ +&maskedFieldNames=password,token +``` + +#### 4.1.3 业务流程 (泳道图) + +**Draw.io文件**: [敏感字段屏蔽_流程图.drawio](敏感字段屏蔽_流程图.drawio) - "下载功能泳道图"页签 + +![下载功能泳道图](敏感字段屏蔽_流程图.drawio) + +```plantuml +@startuml +|用户| +start +:发起下载请求\n携带maskedFieldNames; + +|Gateway| +:接收请求; +:路由到PublicService; + +|FsRestfulApi| +:解析请求参数; +:提取maskedFieldNames; + +if (maskedFieldNames为空?) then (是) + :执行原有下载流程; + |FileSource| + :读取Dolphin文件; + :流式写入Writer; +else (否) + :解析屏蔽字段列表; + note right + maskedFields = + maskedFieldNames + .toLowerCase() + .split(",") + .toSet + end note + + |FileSource| + :调用fileSource.collect()\n收集完整数据; + + |FsRestfulApi| + :调用filterMaskedFieldsFromMetadata()\n过滤元数据; + :调用removeFieldsFromContent()\n移除数据列; + + :创建Writer\n(CSV/Excel); + :写入过滤后的数据; +endif + +|响应| +:返回下载文件流; + +|用户| +:接收文件\n(不含敏感字段); +stop + +@enduml +``` + +#### 4.1.4 核心处理逻辑 (实际实现) + +**实际实现比原设计更简洁**: + +```java +// FsRestfulApi.java - resultsetToExcel方法 (实际实现) + +public void resultsetToExcel( + HttpServletRequest req, + HttpServletResponse response, + @RequestParam(value = "path", required = false) String path, + @RequestParam(value = "outputFileType", defaultValue = "csv") String outputFileType, + @RequestParam(value = "maskedFieldNames", required = false) String maskedFieldNames, // ✅ 新增 + // ... 其他参数 +) { + + // 1. 权限验证 + String userName = ModuleUserUtils.getOperationUser(req); + checkIsUsersDirectory(path, userName); + + // 2. 获取文件系统 + FileSystem fs = fsService.getFileSystemForRead(userName, fsPath); + FileSource fileSource = FileSource.create(fsPath, fs); + + // 3. 创建Writer (根据outputFileType) + FsWriter fsWriter = createWriter(outputFileType, response.getOutputStream(), ...); + + // 4. ✅ 核心逻辑：使用ResultUtils统一处理 + if (StringUtils.isNotBlank(maskedFieldNames)) { + // 使用工具类处理字段屏蔽 + ResultUtils.dealMaskedField(maskedFieldNames, fsWriter, fileSource); + } else { + // 原有流式写入逻辑 + fileSource.write(fsWriter); + } + + // 5. 资源清理 + IOUtils.closeQuietly(fsWriter); + IOUtils.closeQuietly(fileSource); +} +``` + +**关键改进点**: +1. ✅ **简洁性**: 使用`ResultUtils.dealMaskedField()`一行代码替代原来的几十行 +2. ✅ **复用性**: 字段过滤逻辑完全复用，无重复代码 +3. ✅ **可维护性**: 修改过滤逻辑只需修改ResultUtils +4. ✅ **一致性**: 与Pipeline引擎使用相同的工具类，保证行为一致 + +**resultsetsToExcel方法实现**: + +```java +// FsRestfulApi.java - resultsetsToExcel方法 (实际实现) + +public void resultsetsToExcel( + HttpServletRequest req, + HttpServletResponse response, + @RequestParam(value = "path", required = false) String path, + @RequestParam(value = "maskedFieldNames", required = false) String maskedFieldNames, // ✅ 新增 + // ... 其他参数 +) { + + // 1-2. 权限验证和文件系统初始化 (同上) + // ... + + // 3. 创建多结果集Writer + StorageMultiExcelWriter multiExcelWriter = new StorageMultiExcelWriter(outputStream, autoFormat); + + // 4. ✅ 使用ResultUtils统一处理 + if (StringUtils.isNotBlank(maskedFieldNames)) { + ResultUtils.dealMaskedField(maskedFieldNames, multiExcelWriter, fileSource); + } else { + fileSource.write(multiExcelWriter); + } + + // 5. 资源清理 + // ... +} +``` + +--- + +### 4.2 Pipeline导出功能增强设计 + +#### 4.2.1 模块说明 + +**模块路径**: `linkis-engineconn-plugins/pipeline` +**核心类**: +- `PipelineEngineConnExecutor` (语法解析) +- `CSVExecutor` (CSV导出) +- `ExcelExecutor` (Excel导出) + +#### 4.2.2 Pipeline语法扩展 + +**原语法**: +``` +from <源路径> to <目标路径> +``` + +**新语法**: +``` +from <源路径> to <目标路径> without "<字段1,字段2,...>" +``` + +**语法规则**: +1. `without`关键字大小写不敏感 +2. 字段列表必须用**双引号**包裹 +3. 多个字段用逗号分隔 +4. 字段名匹配不区分大小写 + +**示例**: +```scala +// 示例1: 屏蔽单个字段 +from /user/result.dolphin to /export/file.csv without "password" + +// 示例2: 屏蔽多个字段 +from /user/result.dolphin to /export/users.xlsx without "password,apikey,credit_card" + +// 示例3: 向后兼容 +from /user/result.dolphin to /export/file.csv +``` + +#### 4.2.3 正则解析设计 + +```scala +// PipelineEngineConnExecutor.scala + +// 新增正则：支持without子句 +val regexWithMask = + "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s+without\\s+\"([^\"]+)\"\\s*".r + +// 原有正则：不带without +val regexNormal = + "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s*".r +``` + +**正则组成说明**: + +| 部分 | 说明 | 匹配内容 | +|-----|------|---------| +| `(?i)` | 大小写不敏感标志 | - | +| `\\s*from\\s+` | from关键字 | "from " | +| `(\\S+)` | 第1组：源路径 | "/user/result.dolphin" | +| `\\s+to\\s+` | to关键字 | " to " | +| `(\\S+)` | 第2组：目标路径 | "/export/file.csv" | +| `\\s+without\\s+` | without关键字 | " without " | +| `\"([^\"]+)\"` | 第3组：屏蔽字段 | "password,apikey" | + +#### 4.2.4 业务流程 (泳道图) + +**Draw.io文件**: [敏感字段屏蔽_流程图.drawio](敏感字段屏蔽_流程图.drawio) - "导出功能泳道图"页签 + +![导出功能泳道图](敏感字段屏蔽_流程图.drawio) + +```plantuml +@startuml +|前端系统| +start +:生成Pipeline代码\nfrom ... to ... without "..."; +note right + 由其他团队负责 + 根据用户输入生成 +end note +:提交Pipeline任务; + +|PipelineEngineConnExecutor| +:接收Pipeline代码; +:正则匹配解析; + +if (匹配regexWithMask?) then (是) + :提取sourcePath; + :提取destPath; + :提取maskedFields; + note right + maskedFields = + "password,apikey" + end note + + :将maskedFields放入options; + note right + options.put( + "pipeline.masked.field.names", + maskedFields + ) + end note + +else (否，匹配regexNormal?) + :提取sourcePath; + :提取destPath; + :options不含屏蔽字段; +endif + +:根据目标文件扩展名\n选择执行器; + +|CSVExecutor/ExcelExecutor| +:从options获取\npipeline.masked.field.names; + +if (maskedFieldNames不为空?) then (是) + :解析屏蔽字段列表; + + |FileSource| + :调用fileSource.collect()\n收集完整数据; + + |Executor| + :计算保留的列索引; + :过滤元数据; + :过滤数据内容; + :写入目标文件; + +else (否) + |FileSource| + :流式写入\n(原逻辑); +endif + +|文件系统| +:写入导出文件\n到目标路径; + +|响应| +:返回执行成功; + +stop +@enduml +``` + +#### 4.2.5 时序图 (详细代码流程) + +**Draw.io文件**: [敏感字段屏蔽_时序图.drawio](敏感字段屏蔽_时序图.drawio) - "下载功能时序图"和"导出功能时序图"页签 + +![下载功能时序图](敏感字段屏蔽_时序图.drawio) +![导出功能时序图](敏感字段屏蔽_时序图.drawio) + +```plantuml +@startuml +participant "Entrance" as Entrance +participant "PipelineEngineConn\nExecutor" as Executor +participant "PipelineExecutor\nSelector" as Selector +participant "CSVExecutor" as CSV +participant "FileSource" as Source +participant "CSVFsWriter" as Writer +participant "FileSystem" as FS + +Entrance -> Executor: executeLine(code) +note right + code = "from /a/b.dolphin + to /c/d.csv + without \"password,apikey\"" +end note + +Executor -> Executor: 正则匹配 +activate Executor + +alt 匹配regexWithMask + Executor -> Executor: 提取(sourcePath, destPath, maskedFields) + Executor -> Executor: enhancedOptions.put(\n "pipeline.masked.field.names",\n maskedFields) +else 匹配regexNormal + Executor -> Executor: 提取(sourcePath, destPath) + note right: options不含屏蔽字段 +else 语法错误 + Executor --> Entrance: 抛出PipeLineErrorException +end + +Executor -> Selector: select(sourcePath, destPath, enhancedOptions) +Selector -> Selector: getSuffix(destPath) +note right: ".csv" -> CSVExecutor + +Selector -> CSV: 创建CSVExecutor实例 +Selector -> CSV: init(enhancedOptions) +CSV -> CSV: 保存options + +Selector --> Executor: 返回CSVExecutor + +Executor -> CSV: execute(sourcePath, destPath, context) +deactivate Executor + +activate CSV +CSV -> CSV: 从options获取\nmaskedFieldNames + +alt maskedFieldNames不为空 + CSV -> Source: FileSource.create(sourcePath, fs) + Source --> CSV: 返回fileSource + + CSV -> Source: fileSource.collect() + activate Source + Source -> Source: 读取Dolphin文件 + Source --> CSV: 返回collectedData\n(metadata + content) + deactivate Source + + CSV -> CSV: 解析maskedFields =\nmaskedFieldNames\n .split(",")\n .toSet + + CSV -> CSV: filterAndWriteData(\n collectedData,\n maskedFields,\n csvWriter) + activate CSV + + CSV -> CSV: 计算retainedIndices + note right + retainedIndices = + metadata.zipWithIndex + .filter(col => + !maskedFields.contains( + col.columnName)) + .map(_._2) + end note + + CSV -> CSV: 过滤metadata + CSV -> Writer: addMetaData(filteredMetadata) + + loop 遍历每行数据 + CSV -> CSV: 过滤row数据 + CSV -> Writer: addRecord(filteredRow) + end + + CSV -> Writer: flush() + deactivate CSV + +else maskedFieldNames为空 + CSV -> Source: fileSource.write(csvWriter) + note right: 原流式写入逻辑 +end + +CSV -> Writer: close() +CSV -> Source: close() +CSV -> FS: close() + +CSV --> Executor: 返回ExecuteResponse +deactivate CSV + +Executor --> Entrance: 返回成功 +@enduml +``` + +#### 4.2.6 CSVExecutor核心代码 (实际实现) + +**实际实现更简洁**: + +```scala +// CSVExecutor.scala (实际实现) + +class CSVExecutor extends PipeLineExecutor { + + private var options: util.Map[String, String] = _ + + override def init(options: util.Map[String, String]): Unit = { + this.options = options + } + + override def execute( + sourcePath: String, + destPath: String, + engineExecutionContext: EngineExecutionContext + ): ExecuteResponse = { + + // 1. ✅ 获取屏蔽字段参数 (从PipelineEngineConnExecutor传入) + val maskedFieldNames = options.getOrDefault("pipeline.masked.field.names", "") + + // 2. 验证源文件 + if (!sourcePath.contains(STORAGE_RS_FILE_SUFFIX.getValue)) { + throw new PipeLineErrorException(...) + } + if (!FileSource.isResultSet(sourcePath)) { + throw new PipeLineErrorException(...) + } + + // 3. 创建文件系统 + val sourceFsPath = new FsPath(sourcePath) + val destFsPath = new FsPath(destPath) + val sourceFs = FSFactory.getFs(sourceFsPath) + sourceFs.init(null) + val destFs = FSFactory.getFs(destFsPath) + destFs.init(null) + + try { + // 4. 创建FileSource + val fileSource = FileSource.create(sourceFsPath, sourceFs) + if (!FileSource.isTableResultSet(fileSource)) { + throw new PipeLineErrorException(...) + } + + // 5. 获取配置参数 + var nullValue = options.getOrDefault(PIPELINE_OUTPUT_SHUFFLE_NULL_TYPE, "NULL") + if (BLANK.equalsIgnoreCase(nullValue)) nullValue = "" + + // 6. 创建输出流和Writer + val outputStream = destFs.write(destFsPath, PIPELINE_OUTPUT_ISOVERWRITE_SWITCH.getValue(options)) + OutputStreamCache.osCache.put(engineExecutionContext.getJobId.get, outputStream) + + val cSVFsWriter = CSVFsWriter.getCSVFSWriter( + PIPELINE_OUTPUT_CHARSET_STR.getValue(options), + PIPELINE_FIELD_SPLIT_STR.getValue(options), + PIPELINE_FIELD_QUOTE_RETOUCH_ENABLE.getValue(options), + outputStream + ) + + try { + // 7. ✅ 核心逻辑：使用ResultUtils统一处理 + if (StringUtils.isNotBlank(maskedFieldNames)) { + logger.info(s"Applying field masking: $maskedFieldNames") + // 使用工具类处理字段屏蔽 + ResultUtils.dealMaskedField(maskedFieldNames, cSVFsWriter, fileSource) + } else { + // 原有流式写入逻辑 + logger.info("No field masking, using stream write") + fileSource.addParams("nullValue", nullValue).write(cSVFsWriter) + } + } finally { + IOUtils.closeQuietly(cSVFsWriter) + IOUtils.closeQuietly(fileSource) + } + } finally { + IOUtils.closeQuietly(sourceFs) + IOUtils.closeQuietly(destFs) + } + + super.execute(sourcePath, destPath, engineExecutionContext) + } +} +``` + +**关键改进点**: +1. ✅ **简化实现**: 使用`ResultUtils.dealMaskedField()`替代原设计中的`filterAndWriteData()`方法 +2. ✅ **代码复用**: 与FsRestfulApi共享相同的字段过滤逻辑 +3. ✅ **无需自实现**: 删除了原设计中的`filterAndWriteData()`, `filterRow()`等辅助方法 +4. ✅ **更好的架构**: 字段过滤逻辑集中在Storage层，符合分层架构原则 + +**ExcelExecutor实现**: + +```scala +// ExcelExecutor.scala (实际实现，与CSVExecutor类似) + +class ExcelExecutor extends PipeLineExecutor { + override def execute(...): ExecuteResponse = { + val maskedFieldNames = options.getOrDefault("pipeline.masked.field.names", "") + + // ... 初始化代码 ... + + if (StringUtils.isNotBlank(maskedFieldNames)) { + // 使用ResultUtils处理字段屏蔽 + ResultUtils.dealMaskedField(maskedFieldNames, excelFsWriter, fileSource) + } else { + fileSource.addParams("nullValue", nullValue).write(excelFsWriter) + } + } +} +``` + +--- + +## 5. 数据结构/存储设计 + +### 5.1 Dolphin文件格式 + +**文件结构**: + +``` ++-------------------+ +| Magic Header (7B) | "dolphin" ++-------------------+ +| Type Flag (10B) | "TABLE " (固定10字节) ++-------------------+ +| Metadata Length | 元数据区长度 ++-------------------+ +| Metadata | 列定义JSON +| { | +| columns: [ | +| { | +| columnName, | +| dataType, | +| comment | +| } | +| ] | +| } | ++-------------------+ +| Data Records | 行数据 +| Row 1 | 字段1,字段2,... +| Row 2 | +| ... | ++-------------------+ +``` + +### 5.2 内存数据结构 + +#### 5.2.1 元数据结构 + +```java +// 元数据数组 +Map[] metadata = { + { + "columnName": "id", + "dataType": "int", + "comment": "用户ID" + }, + { + "columnName": "password", + "dataType": "string", + "comment": "密码" // 需要屏蔽 + }, + { + "columnName": "email", + "dataType": "string", + "comment": "邮箱" + } +} +``` + +#### 5.2.2 数据内容结构 + +```java +// 数据行数组 +List fileContent = [ + ["1", "pwd123", "alice@example.com"], + ["2", "secret456", "bob@example.com"] +] +``` + +#### 5.2.3 过滤后结构 + +```java +// 过滤后元数据 (移除password) +Map[] filteredMetadata = { + { + "columnName": "id", + "dataType": "int", + "comment": "用户ID" + }, + { + "columnName": "email", + "dataType": "string", + "comment": "邮箱" + } +} + +// 过滤后数据 (移除password列) +List filteredContent = [ + ["1", "alice@example.com"], + ["2", "bob@example.com"] +] +``` + +### 5.3 配置数据 + +#### 5.3.1 新增配置项 + +```properties +# Pipeline导出行数限制 (方案A内存保护) +pipeline.export.max.rows=100000 + +# 内存检查开关 +pipeline.export.memory.check.enabled=true + +# 内存使用阈值 +pipeline.export.memory.threshold=0.8 +``` + +--- + +## 6. 接口设计 + +接口设计文档已录入API DESIGN系统： +http://apidesign.weoa.com + +### 6.1 接口清单 + +| 接口路径 | 方法 | 说明 | 变更类型 | +|---------|------|------|---------| +| `/api/rest_j/v1/filesystem/resultsetToExcel` | GET | 单结果集下载 | 参数扩展 | +| `/api/rest_j/v1/filesystem/resultsetsToExcel` | GET | 多结果集下载 | 参数扩展 | + +### 6.2 参数说明 + +#### 新增参数 + +| 参数名 | 类型 | 必填 | 默认值 | 说明 | 示例 | +|-------|------|------|-------|------|------| +| maskedFieldNames | String | 否 | null | 屏蔽字段列表，逗号分隔，不区分大小写 | password,apikey,ssn | + +#### 响应说明 + +**成功响应**: 返回文件流 (与原接口一致) + +**错误响应**: +```json +{ + "status": 1, + "message": "字段名格式错误", + "data": null +} +``` + +--- + +## 7. 专利点识别 + +### 7.1 潜在专利点 + +#### 专利点1: 基于Pipeline语法扩展的字段级数据脱敏方法 + +**技术特点**: +1. 通过扩展Pipeline DSL语法，实现声明式字段屏蔽 +2. 在数据导出过程中动态解析语法并应用屏蔽逻辑 +3. 保持向后兼容的同时提供灵活的字段控制能力 + +**创新点**: +- 使用正则匹配提取屏蔽字段，避免修改前端代码 +- 基于语法扩展的声明式安全控制 +- 职责分离：前端生成、后端执行 + +#### 专利点2: 基于内存感知的字段屏蔽策略选择方法 + +**技术特点**: +1. 根据结果集大小和内存情况动态选择处理策略 +2. 小结果集使用collect模式，大结果集使用流式模式或拒绝 +3. 内存阈值检查机制，防止OOM + +**创新点**: +- 自适应的字段屏蔽策略 +- 基于运行时内存监控的容量保护 +- 性能与安全的平衡 + +### 7.2 专利录入 + +专利信息已录入到"BDP 专利"文档： +http://docs.weoa.com/sheets/2wAlXOo1WBHwPrAP/zDmhC + +--- + +## 8. 附录 + +### 8.1 关键文件清单 + +| 文件路径 | 说明 | 修改类型 | +|---------|------|---------| +| `linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/utils/ResultUtils.java` | ✅ 工具类 (新增189行) | **核心改进** - 提取公共字段过滤逻辑 | +| `linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java` | REST API | 参数扩展+调用ResultUtils | +| `linkis-engineconn-plugins/pipeline/src/main/scala/org/apache/linkis/manager/engineplugin/pipeline/executor/PipelineEngineConnExecutor.scala` | Pipeline执行器 | 正则扩展 | +| `linkis-engineconn-plugins/pipeline/src/main/scala/org/apache/linkis/manager/engineplugin/pipeline/executor/CSVExecutor.scala` | CSV导出 | 调用ResultUtils | +| `linkis-engineconn-plugins/pipeline/src/main/scala/org/apache/linkis/manager/engineplugin/pipeline/executor/ExcelExecutor.scala` | Excel导出 | 调用ResultUtils | + +### 8.2 配置项清单 + +| 配置项 | 默认值 | 说明 | 模块 | +|-------|-------|------|------| +| `wds.linkis.workspace.resultset.download.maxsize.csv` | 5000 | CSV下载最大行数 | 下载 | +| `wds.linkis.workspace.resultset.download.maxsize.excel` | 5000 | Excel下载最大行数 | 下载 | +| `pipeline.export.max.rows` | 100000 | Pipeline导出最大行数 | 导出 | +| `pipeline.export.memory.check.enabled` | true | 是否启用内存检查 | 导出 | +| `pipeline.export.memory.threshold` | 0.8 | 内存使用阈值 | 导出 | + +### 8.3 测试用例清单 + +#### 功能测试 + +| 用例ID | 用例名称 | 优先级 | +|-------|---------|--------| +| TC001 | 下载单字段屏蔽-CSV | P0 | +| TC002 | 下载多字段屏蔽-Excel | P0 | +| TC003 | 导出Pipeline语法-单字段 | P0 | +| TC004 | 导出Pipeline语法-多字段 | P0 | +| TC005 | 向后兼容-不传参数 | P0 | +| TC006 | 字段名大小写不敏感 | P1 | +| TC007 | 屏蔽不存在字段 | P1 | +| TC008 | 屏蔽所有字段 | P2 | + +#### 性能测试 + +| 用例ID | 数据量 | 屏蔽字段数 | 期望性能 | +|-------|-------|----------|---------| +| PT001 | 1万行×10列 | 2 | <3秒 | +| PT002 | 5万行×50列 | 5 | <8秒 | +| PT003 | 10万行×100列 | 10 | <15秒 | + +--- + +## 9. 性能优化策略 + +### 9.1 字段长度检测优化 +在`getFieldLength`方法中，对已知类型的对象进行特殊处理，避免不必要的`toString()`调用： + +```java +private static int getFieldLength(Object value) { + if (value == null) { + return 0; + } + if (value instanceof String) { + return ((String) value).length(); + } + return value.toString().length(); +} +``` + +### 9.2 大结果集处理优化 +对于大结果集，采用分批处理策略： +1. 设置内存使用阈值监控 +2. 超过阈值时采用流式处理 +3. 提供处理进度反馈机制 + +### 9.3 缓存机制优化 +在字段处理过程中使用Set来存储已处理的字段名，避免重复处理： + +```java +// 使用Set来存储已经处理过的字段名，避免重复处理 +Set processedFields = new HashSet<>(); +``` + +--- + +## 10. 变更历史 + +| 版本 | 日期 | 变更内容 | 作者 | +|-----|------|---------|------| +| v1.0 | 2025-10-28 | 初始版本 - 完成系统设计 | Claude Code | +| v1.1 | 2025-10-30 | ✅ 实现完成 - 更新实际实现细节，添加ResultUtils工具类设计 | 开发团队 | + +**v1.1版本主要变更**: +1. 新增ResultUtils工具类设计章节 (3.5节) +2. 更新PublicService实现代码，反映实际使用ResultUtils的简化实现 (4.1.4节) +3. 更新Pipeline引擎实现代码，反映实际使用ResultUtils的简化实现 (4.2.6节) +4. 更新文件清单，突出ResultUtils核心地位 (9.1节) +5. 添加实施总结章节，说明架构改进点 + +--- + +**文档结束** \ No newline at end of file diff --git a/docs/1.17.0/requirements/resultset-field-truncation.md b/docs/1.17.0/requirements/resultset-field-truncation.md new file mode 100644 index 0000000000..9ef5c0b075 --- /dev/null +++ b/docs/1.17.0/requirements/resultset-field-truncation.md @@ -0,0 +1,230 @@ +# 结果集查看、下载和导出接口优化需求文档 + +## 文档信息 +| 项目 | 信息 | +|-----|------| +| 文档版本 | v1.1 (已实现) | +| 创建日期 | 2025-10-27 | +| 更新日期 | 2025-10-30 | +| 当前版本 | Linkis 1.17.0 | +| 负责模块 | linkis-pes-publicservice + pipeline + linkis-storage | +| 开发分支 | feature/1.17.0-resultset-field-masking | +| 状态 | ✅ 开发完成，已测试 | + +## 实施总结 + +### 代码修改统计 +本次开发包含**敏感字段屏蔽**和**字段截取**两个功能： + +```bash +15 files changed, 4166 insertions(+), 386 deletions(-) +``` + +### 新增文件 + +| 文件 | 行数 | 说明 | +|------|-----|------| +| `ResultUtils.java` | 514行 | 核心工具类，包含字段屏蔽和截取逻辑 | +| `FieldTruncationResult.java` | 73行 | 字段截取结果封装实体类 | +| `OversizedFieldInfo.java` | 68行 | 超长字段信息实体类 | + +### 修改文件 + +| 文件 | 修改类型 | 说明 | +|------|---------|------| +| `LinkisStorageConf.scala` | 配置扩展 (+11行) | 新增字段截取相关配置项 | +| `WorkSpaceConfiguration.java` | 配置扩展 (+4行) | 新增功能开关配置 | +| `FsRestfulApi.java` | 功能增强 (218改动) | 下载接口支持字段屏蔽和截取 | +| `PipelineEngineConnExecutor.scala` | 语法扩展 (+16改动) | 支持without和truncate子句 | +| `CSVExecutor.scala` | 功能增强 (70改动) | CSV导出支持屏蔽和截取 | +| `ExcelExecutor.scala` | 功能增强 (140改动) | Excel导出支持屏蔽和截取 | +| 文档 | 新增4份 | 需求和设计文档 | + +### 核心改进点 + +1. **统一工具类**: 将字段屏蔽和截取逻辑提取到`ResultUtils`工具类，实现代码复用 +2. **组合功能**: 支持字段屏蔽和字段截取同时使用（`applyFieldMaskingAndTruncation`方法） +3. **可配置化**: 所有阈值和开关都通过`CommonVars`配置管理 +4. **向后兼容**: 功能可选，不影响现有功能 +5. **标记机制**: 截取后的字段会在列名添加`(truncated to N chars)`后缀标记 + +### 实现的核心方法 + +**ResultUtils工具类方法**: +- `detectAndHandle()`: 检测并处理超长字段（主入口方法） +- `detectOversizedFields()`: 检测超长字段，返回超长字段列表 +- `truncateFields()`: 截取超长字段值 +- `applyFieldMaskingAndTruncation()`: 同时应用字段屏蔽和截取 + +--- + +## 1. 需求概述 + +### 1.1 需求主题 +结果集查看、下载和导出接口优化 - 超长字段截取功能 + +### 1.2 需求背景 +当前结果集查看功能存在以下问题： +- 当某一列字段内容超过10000字符时，会导致结果集无法正常查看 +- 缺少对超长字段的检测和处理机制 +- 用户无法获知哪些字段超长，也无法选择处理方式 + +### 1.3 需求目标 +为结果集查看、下载、导出功能增加超长字段检测和截取能力，提升系统稳定性和用户体验。 + +## 2. 功能需求 + +### 2.1 核心功能点 + +#### 2.1.1 结果集查看功能优化 +- **触发条件**：结果集中存在字段值长度超过10000字符 +- **处理逻辑**： + 1. 检测所有字段值长度 + 2. 收集超过10000字符的字段信息（字段名、行号、实际长度） + 3. 最多收集20个超长字段 + 4. 返回超长字段列表给前端，由用户确认是否截取 + 5. 若用户确认截取，则截取前10000个字符后返回结果集 + 6. 若用户取消，则返回原始数据（可能导致查看失败） + +#### 2.1.2 结果集下载功能优化 +- **触发条件**：结果集中存在字段值长度超过10000字符 +- **处理逻辑**：与查看功能相同 + 1. 检测所有字段值长度 + 2. 收集超过10000字符的字段信息 + 3. 最多收集20个超长字段 + 4. 返回超长字段列表给前端确认 + 5. 若用户确认截取，则截取前10000个字符后下载 + 6. 若用户取消，则下载原始数据 + +#### 2.1.3 结果集导出功能优化 +- **触发条件**：结果集中存在字段值长度超过32767字符 +- **处理逻辑**： + 1. 检测所有字段值长度 + 2. 收集超过32767字符的字段信息（字段名、行号、实际长度） + 3. 最多收集20个超长字段 + 4. 返回超长字段列表给前端确认 + 5. 若用户确认截取，则截取前32767个字符后导出 + 6. 若用户取消，则导出原始数据（可能导致导出失败） + +### 2.2 功能约束 + +#### 2.2.1 超长字段收集上限 +- 最多收集20个超长字段信息 +- 超过20个时，只返回前20个 + +#### 2.2.2 截取长度配置 +- 查看和下载：默认10000字符，可配置 +- 导出：默认32767字符，可配置 + +#### 2.2.3 功能开关 +- 必须提供功能总开关，关闭时相当于回退到原版本功能 +- 开关关闭时，不进行任何检测和截取 + +## 3. 非功能需求 + +### 3.1 性能要求 +- 字段长度检测不应显著增加接口响应时间 +- 对于大结果集，检测逻辑应高效执行 + +### 3.2 兼容性要求 +- 遵循最小改动原则，不影响现有功能 +- 功能开关关闭时，行为与原版本完全一致 + +### 3.3 可配置性要求 +- 所有阈值参数必须可配置 +- 配置必须使用 `CommonVars` 统一管理 +- 参考 `JobhistoryConfiguration` 的配置方式 + +## 4. 接口设计要求 + +### 4.1 返回数据结构 +需要在结果集相关接口的响应中增加以下信息： + +```json +{ + "hasOversizedFields": true, + "oversizedFields": [ + { + "fieldName": "column1", + "rowIndex": 0, + "actualLength": 15000, + "maxLength": 10000 + } + ], + "maxOversizedFieldCount": 20, + "data": "结果集数据" +} +``` + +### 4.2 前端交互流程 +1. 后端检测到超长字段，返回超长字段列表 +2. 前端展示提示弹窗，显示超长字段信息 +3. 用户选择是否截取 +4. 前端带着用户选择结果重新请求接口 +5. 后端根据用户选择返回截取或原始数据 + +## 5. 配置项清单 + +| 配置项名称 | 默认值 | 说明 | +|-----------|--------|------| +| `linkis.resultset.field.truncation.enabled` | `false` | 功能总开关 | +| `linkis.resultset.field.view.max.length` | `10000` | 查看功能字段最大长度 | +| `linkis.resultset.field.download.max.length` | `10000` | 下载功能字段最大长度 | +| `linkis.resultset.field.export.max.length` | `32767` | 导出功能字段最大长度 | +| `linkis.resultset.field.oversized.max.count` | `20` | 最多收集超长字段数量 | + +## 6. 实施范围 + +### 6.1 开发范围 +- 仅实现后端接口功能 +- 不涉及前端页面开发 + +### 6.2 代码边界 +- 不修改现有表结构 +- 不引入新的第三方依赖 +- 不修改现有公共接口签名（只扩展返回数据） + +## 7. 验收标准 + +### 7.1 功能验收 +- [x] ✅ 功能开关关闭时，行为与原版本一致 +- [x] ✅ 功能开关开启时，能正确检测超长字段 +- [x] ✅ 能返回正确的超长字段信息列表（通过FieldTruncationResult封装） +- [x] ✅ 用户选择截取时，能正确截取指定长度 +- [x] ✅ 超长字段超过20个时，只返回前20个 +- [x] ✅ 截取后的字段会在列名添加标记`(truncated to N chars)` + +### 7.2 配置验收 +- [x] ✅ 所有配置项使用 `CommonVars` 管理 +- [x] ✅ 配置项放在对应模块的 Configuration 类中（LinkisStorageConf和WorkSpaceConfiguration） +- [x] ✅ 配置项可以正确读取和生效 + +### 7.3 兼容性验收 +- [x] ✅ 不影响现有结果集查看功能 +- [x] ✅ 不影响现有结果集下载功能 +- [x] ✅ 不影响现有结果集导出功能 + +### 7.4 扩展功能验收 (新增) +- [x] ✅ 支持字段屏蔽和字段截取同时使用 +- [x] ✅ Pipeline语法支持truncate参数 +- [x] ✅ CSV和Excel导出都支持字段截取 + +## 8. 风险评估 + +### 8.1 技术风险 +- **性能影响**：字段长度检测可能影响性能，需要优化检测逻辑 +- **内存占用**：大结果集检测可能增加内存占用 + +### 8.2 兼容性风险 +- **前端兼容**：老版本前端不识别新增的返回字段，需要保证向下兼容 + +## 9. 参考资料 + +### 9.1 相关代码模块 +- 结果集查看相关代码 +- 结果集下载相关代码 +- 结果集导出相关代码 + +### 9.2 配置参考 +- `org.apache.linkis.jobhistory.conf.JobhistoryConfiguration` +- `org.apache.linkis.common.conf.CommonVars` diff --git a/docs/1.17.0/requirements/resultset-sensitive-field-masking.md b/docs/1.17.0/requirements/resultset-sensitive-field-masking.md new file mode 100644 index 0000000000..1c29f87b40 --- /dev/null +++ b/docs/1.17.0/requirements/resultset-sensitive-field-masking.md @@ -0,0 +1,1182 @@ +# Linkis结果集下载和导出功能支持敏感字段屏蔽需求文档 + +## 文档信息 + +| 项目 | 信息 | +|-----|------| +| 文档版本 | v1.1 (已实现) | +| 创建日期 | 2025-10-27 | +| 更新日期 | 2025-10-30 | +| 当前版本 | Linkis 1.17.0 | +| 负责模块 | linkis-pes-publicservice (Filesystem) + pipeline | +| 开发分支 | feature/1.17.0-resultset-sensitive-field-masking | +| 状态 | ✅ 开发完成，已测试 | + +--- + +## 实施总结 + +### 代码修改统计 + +```bash +7 files changed, 2698 insertions(+), 163 deletions(-) +``` + +| 文件 | 修改类型 | 说明 | +|------|---------|------| +| `FsRestfulApi.java` | 功能增强 | 添加maskedFieldNames参数支持 | +| `ResultUtils.java` | 新增工具类 | 提取公共字段过滤逻辑 | +| `PipelineEngineConnExecutor.scala` | 语法扩展 | 支持without子句解析 | +| `CSVExecutor.scala` | 功能增强 | 实现CSV导出字段屏蔽 | +| `ExcelExecutor.scala` | 功能增强 | 实现Excel导出字段屏蔽 | +| `resultset-sensitive-field-masking.md` | 新增文档 | 需求文档 | +| `resultset-sensitive-field-masking-design.md` | 新增文档 | 设计文档 | + +### 核心改进点 + +1. **代码复用**: 将字段过滤逻辑提取到`ResultUtils`工具类，实现Java和Scala代码共享 +2. **简化实现**: 使用`ResultUtils.dealMaskedField()`统一处理字段屏蔽，减少重复代码 +3. **更好的架构**: 将通用逻辑放在`linkis-storage`模块，提高可维护性 + +--- + +## 1. 需求背景 + +### 1.1 现状说明 + +Linkis当前在结果集查看功能(`/api/rest_j/v1/filesystem/openFile`)中已经实现了敏感字段屏蔽机制,通过`maskedFieldNames`参数支持动态指定需要屏蔽的字段列表。该功能在前端展示结果集时可以有效保护敏感数据。 + +### 1.2 存在的安全风险 + +虽然结果集查看时支持屏蔽敏感字段,但用户仍然可以通过以下接口**绕过屏蔽机制**获取完整的敏感数据: + +1. **单结果集下载接口**: `/api/rest_j/v1/filesystem/resultsetToExcel` +2. **多结果集下载接口**: `/api/rest_j/v1/filesystem/resultsetsToExcel` +3. **其他导出接口**(如存在) + +这导致敏感字段屏蔽功能形同虚设,存在数据泄露风险。 + +### 1.3 需求来源 + +- 数据安全合规要求 +- 敏感信息保护策略的全链路落地 +- 用户权限管理的完善 + +--- + +## 2. 功能现状分析 + +### 2.1 结果集查看功能 (已支持屏蔽) + +#### 接口信息 + +``` +接口路径: /api/rest_j/v1/filesystem/openFile +请求方法: GET +Controller类: org.apache.linkis.filesystem.restful.api.FsRestfulApi +实现方法: openFile() (行625-777) +``` + +#### 敏感字段屏蔽参数 + +| 参数名 | 类型 | 是否必填 | 说明 | 示例 | +|-------|------|---------|------|------| +| maskedFieldNames | String | 否 | 需要屏蔽的字段名,多个字段用逗号分隔(不区分大小写) | password,apikey,secret_token | + +#### 屏蔽实现机制 + +**实现位置**: FsRestfulApi.java 行735-858 + +```java +// 1. 解析屏蔽字段列表 +Set maskedFields = + new HashSet<>(Arrays.asList(maskedFieldNames.toLowerCase().split(","))); + +// 2. 过滤元数据 +Map[] metadata = filterMaskedFieldsFromMetadata(resultmap, maskedFields); + +// 3. 移除数据内容中的对应列 +List fileContent = + removeFieldsFromContent(resultmap, result.getSecond(), maskedFields); +``` + +**关键方法**: + +1. `filterMaskedFieldsFromMetadata()` (行841-858): 从元数据中过滤屏蔽字段 +2. `removeFieldsFromContent()` (行787-838): 从内容数据中移除屏蔽字段列 + +**特性**: +- 不区分大小写匹配字段名 +- 从后向前删除列索引,避免索引变化问题 +- 同时处理元数据和内容数据 + +--- + +### 2.2 单结果集下载功能 (不支持屏蔽) + +#### 接口信息 + +``` +接口路径: /api/rest_j/v1/filesystem/resultsetToExcel +请求方法: GET +Controller类: org.apache.linkis.filesystem.restful.api.FsRestfulApi +实现方法: resultsetToExcel() (行972-1084) +``` + +#### 核心参数 + +| 参数名 | 类型 | 默认值 | 说明 | +|-------|------|-------|------| +| path | String | - | 结果集文件路径(必填) | +| outputFileType | String | csv | 导出格式: csv 或 xlsx | +| csvSeparator | String | , | CSV分隔符 | +| outputFileName | String | downloadResultset | 输出文件名 | +| sheetName | String | result | Excel sheet名称 | +| nullValue | String | NULL | null值替换字符串 | +| limit | Integer | 0 | 行数限制(0表示使用配置值) | +| autoFormat | Boolean | false | 是否自动格式化 | +| keepNewline | Boolean | false | 是否保留换行符 | + +#### 实现流程 + +``` +用户请求 → 权限验证 → 文件系统操作 → 格式判断(CSV/XLSX) +→ Writer初始化 → 数据写入 → 响应流输出 +``` + +**问题**: 当前实现直接将完整的结果集数据写入输出流,**没有任何字段过滤或屏蔽逻辑**。 + +--- + +### 2.3 多结果集下载功能 (不支持屏蔽) + +#### 接口信息 + +``` +接口路径: /api/rest_j/v1/filesystem/resultsetsToExcel +请求方法: GET +Controller类: org.apache.linkis.filesystem.restful.api.FsRestfulApi +实现方法: resultsetsToExcel() (行1105-1189) +``` + +#### 核心参数 + +| 参数名 | 类型 | 默认值 | 说明 | +|-------|------|-------|------| +| path | String | - | 结果集目录路径(必填) | +| outputFileName | String | downloadResultset | 输出文件名 | +| nullValue | String | NULL | null值替换字符串 | +| limit | Integer | 0 | 每个结果集的行数限制 | +| autoFormat | Boolean | false | 是否自动格式化 | + +#### 特殊说明 + +- **仅支持XLSX格式** +- path参数为目录路径,包含多个结果集文件 +- 使用`StorageMultiExcelWriter`将多个结果集合并到单个Excel的不同Sheet +- 自动按文件序号排序: `ResultSetUtils.sortByNameNum()` + +**问题**: 与单结果集下载类似,**没有任何字段过滤或屏蔽逻辑**。 + +--- + +### 2.4 结果集导出功能 (不支持屏蔽) + +#### 功能说明 + +结果集导出功能与下载功能**不同**,它使用**Pipeline引擎**将dolphin结果集文件导出到服务器共享目录,而非直接下载到客户端。 + +#### 实现方式 + +**核心引擎**: Pipeline引擎 +**实现语言**: Scala +**代码路径**: `linkis-engineconn-plugins/pipeline/` + +#### 工作流程 + +``` +用户操作流程: +1. 用户在前端点击"导出"按钮 +2. 前端弹出导出配置对话框 (resultsExport.vue) +3. 用户选择: + - 导出文件名 + - 导出格式 (CSV/Excel) + - 目标路径 (服务器共享目录) + - 是否导出全部结果集 (多结果集时) +4. 前端生成Pipeline代码: from <源路径> to <目标路径> +5. 提交Pipeline脚本到引擎执行 +6. Pipeline引擎读取dolphin文件 → 转换格式 → 写入目标目录 +``` + +#### 前端实现 + +**文件路径**: `linkis-web/src/components/consoleComponent/resultsExport.vue` + +**导出配置参数**: + +| 参数 | 类型 | 说明 | 验证规则 | +|-----|------|------|---------| +| name | String | 导出文件名 | 1-200字符,仅支持英文/数字/中文 | +| path | String | 目标目录路径 | 必填,从目录树选择 | +| format | String | 导出格式 | 1=CSV, 2=Excel | +| isAll | Boolean | 是否导出全部结果集 | 仅多结果集且Excel格式时可选 | + +**Pipeline代码生成逻辑**: + +```javascript +// resultsExport.vue 导出确认方法 +exportConfirm() { + // 生成临时脚本名称 + const tabName = `new_stor_${Date.now()}.out`; + + // 确定源路径 + let temPath = this.currentPath; // 当前结果集路径 + if (this.isAll) { + // 导出全部时,源路径为目录(不带文件名) + temPath = temPath.substring(0, temPath.lastIndexOf('/')); + } + + // 根据格式添加扩展名 + const exportOptionName = this.exportOption.format === '2' + ? `${this.exportOption.name}.xlsx` + : `${this.exportOption.name}.csv`; + + // 生成Pipeline执行代码 + const code = `from ${temPath} to ${this.exportOption.path}/${exportOptionName}`; + + // 添加临时脚本并自动执行 + this.dispatch('Workbench:add', { id: md5Path, code, saveAs: true }, (f) => { + this.$nextTick(() => { + this.dispatch('Workbench:run', { id: md5Path }); + }); + }); +} +``` + +#### Pipeline引擎实现 + +**执行入口**: `PipelineEngineConnExecutor.scala` (行69-89) + +```scala +// 正则解析Pipeline语法 +val regex = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s?".r + +code match { + case regex(sourcePath, destPath) => + // 选择合适的执行器 + PipelineExecutorSelector + .select(sourcePath, destPath, options) + .execute(sourcePath, destPath, engineExecutorContext) +} +``` + +**执行器选择逻辑**: `PipelineExecutorSelector.scala` + +```scala +def select(sourcePath: String, destPath: String, options: Map[String, String]): PipeLineExecutor = { + // 根据目标文件扩展名选择执行器 + getSuffix(destPath) match { + case ".csv" => CSVExecutor // CSV导出 + case ".xlsx" => ExcelExecutor // Excel导出 + case _ if sameFileName => CopyExecutor // 文件复制 + case _ => throw UnsupportedOutputTypeException + } +} +``` + +#### 三大执行器实现 + +##### 1. CSVExecutor - CSV格式导出 + +**文件**: `CSVExecutor.scala` + +**执行流程**: + +```scala +override def execute(sourcePath: String, destPath: String, context: EngineExecutionContext): ExecuteResponse = { + // 1. 验证源文件是否为结果集 + if (!FileSource.isResultSet(sourcePath)) { + throw NotAResultSetFileException + } + + // 2. 创建文件系统 + val sourceFs = FSFactory.getFs(new FsPath(sourcePath)) + val destFs = FSFactory.getFs(new FsPath(destPath)) + + // 3. 创建FileSource读取结果集 + val fileSource = FileSource.create(new FsPath(sourcePath), sourceFs) + + // 4. 获取配置参数 + val nullValue = options.getOrDefault("pipeline.output.shuffle.null.type", "NULL") + val charset = options.getOrDefault("pipeline.output.charset", "UTF-8") + val separator = options.getOrDefault("pipeline.field.split", ",") + val quoteRetouchEnable = options.getOrDefault("pipeline.field.quote.retoch.enable", false) + + // 5. 创建CSV Writer + val outputStream = destFs.write(new FsPath(destPath), isOverwrite = true) + val csvWriter = CSVFsWriter.getCSVFSWriter(charset, separator, quoteRetouchEnable, outputStream) + + // 6. 写入数据 (仅处理nullValue参数) + fileSource.addParams("nullValue", nullValue).write(csvWriter) + + // 7. 清理资源 + IOUtils.closeQuietly(csvWriter) + IOUtils.closeQuietly(fileSource) +} +``` + +**问题**: ❌ **没有任何敏感字段屏蔽逻辑** + +##### 2. ExcelExecutor - Excel格式导出 + +**文件**: `ExcelExecutor.scala` + +**执行流程**: + +```scala +override def execute(sourcePath: String, destPath: String, context: EngineExecutionContext): ExecuteResponse = { + val sourceFs = FSFactory.getFs(new FsPath(sourcePath)) + val destFs = FSFactory.getFs(new FsPath(destPath)) + + val outputStream = destFs.write(new FsPath(destPath), isOverwrite = true) + + // 支持两种模式: + // 模式1: 单个结果集文件 (sourcePath包含".") + if (sourcePath.contains(".")) { + val fileSource = FileSource.create(new FsPath(sourcePath), sourceFs) + val excelWriter = ExcelFsWriter.getExcelFsWriter( + charset = "utf-8", + sheetName = "result", + dateFormat = "yyyy-MM-dd HH:mm:ss", + outputStream, + autoFormat = false + ) + fileSource.addParams("nullValue", nullValue).write(excelWriter) + } + // 模式2: 多个结果集 (sourcePath为目录) + else { + val fsPathList = sourceFs.listPathWithError(new FsPath(sourcePath)).getFsPaths + ResultSetUtils.sortByNameNum(fsPathList) // 按序号排序 + val fileSource = FileSource.create(fsPathList.toArray, sourceFs) + val multiExcelWriter = new StorageMultiExcelWriter(outputStream, autoFormat) + fileSource.addParams("nullValue", nullValue).write(multiExcelWriter) + } +} +``` + +**问题**: ❌ **同样没有敏感字段屏蔽逻辑** + +##### 3. CopyExecutor - 文件复制 + +**文件**: `CopyExecutor.scala` + +```scala +override def execute(sourcePath: String, destPath: String, context: EngineExecutionContext): ExecuteResponse = { + val sourceFs = FSFactory.getFs(new FsPath(sourcePath)) + val destFs = FSFactory.getFs(new FsPath(destPath)) + + val inputStream = sourceFs.read(new FsPath(sourcePath)) + val outputStream = destFs.write(new FsPath(destPath), isOverwrite = true) + + // 直接流复制,不做任何处理 + IOUtils.copy(inputStream, outputStream) +} +``` + +**问题**: ❌ **直接复制文件,完全绕过所有检查** + +#### 关键配置项 + +**文件**: `PipelineEngineConfiguration.scala` + +| 配置项 | 默认值 | 说明 | +|-------|-------|------| +| pipeline.output.charset | UTF-8 | 输出字符集 | +| pipeline.field.split | , | CSV字段分隔符 | +| pipeline.output.shuffle.null.type | NULL | 空值替换标记 | +| pipeline.field.quote.retoch.enable | false | 引号处理开关 | +| pipeline.output.isoverwrite | true | 是否覆盖已存在文件 | +| wds.linkis.pipeline.export.excel.auto_format.enable | false | Excel自动格式化 | + +#### Dolphin结果集文件格式 + +**文件**: `Dolphin.scala` + +```scala +object Dolphin { + val MAGIC = "dolphin" // 文件头魔数 (7字节) + val DOLPHIN_FILE_SUFFIX = ".dolphin" // 文件后缀 + val COL_SPLIT = "," // 列分隔符 + val NULL = "NULL" // 空值标记 + val INT_LEN = 10 // 整数字段长度(固定10字节) +} +``` + +**文件结构**: +1. 文件头 (7字节): "dolphin" +2. 类型标识 (10字节): TABLE/PICTURE/TEXT等 +3. 元数据区 (变长): 列名、数据类型、注释等 +4. 数据区 (变长): 按行存储的数据记录 + +#### 结果集读取流程 + +**FileSource.scala** → **ResultsetFileSource.scala** → **StorageCSVWriter.scala** + +```scala +// ResultsetFileSource.scala - 结果集字段处理 +class ResultsetFileSource(fileSplits: Array[FileSplit]) extends AbstractFileSource(fileSplits) { + // 应用shuffle变换 (仅处理NULL值和Double格式) + shuffle({ + case t: TableRecord => + new TableRecord(t.row.map { + case null | "NULL" => + val nullValue = getParams.getOrDefault("nullValue", "NULL") + nullValue + case value: Double => StorageUtils.doubleToString(value) + case rvalue => rvalue + }) + }) +} +``` + +**关键发现**: +- ✅ 有NULL值处理 +- ✅ 有数值格式化 +- ❌ **没有字段级别的过滤或屏蔽** +- ❌ **没有敏感字段检查** +- ❌ **没有数据脱敏处理** + +#### 导出目标路径 + +根据配置项 `wds.linkis.filesystem.root.path`: + +| 文件系统类型 | 默认根路径 | 说明 | +|------------|----------|------| +| 本地文件系统 | file:///tmp/linkis/ | LOCAL_USER_ROOT_PATH | +| HDFS | hdfs:///tmp/{user}/linkis/ | HDFS_USER_ROOT_PATH_PREFIX + user + SUFFIX | + +**用户可选择的导出路径**: +- 个人工作目录 +- 共享目录(如配置允许) +- 项目目录 + +#### 核心安全问题 + +##### 问题1: dolphin源文件包含所有字段 + +``` +执行SQL: SELECT name, password, email FROM users; + ↓ +生成结果集: /user/hadoop/linkis/result_001.dolphin + ↓ +dolphin文件内容: + - 元数据: [name, password, email] + - 数据: ["Alice", "pwd123", "alice@example.com"] + ["Bob", "secret456", "bob@example.com"] +``` + +**问题**: 结果集文件已包含所有敏感字段 + +##### 问题2: 导出时未进行字段屏蔽 + +``` +用户执行导出: + from /user/hadoop/linkis/result_001.dolphin + to /shared/exports/users.csv + ↓ +CSVExecutor.execute() 流程: + 1. FileSource.create() - 读取dolphin文件 + 2. fileSource.addParams("nullValue", "NULL") + 3. fileSource.write(csvWriter) + ↓ +输出文件 /shared/exports/users.csv: + name,password,email + Alice,pwd123,alice@example.com + Bob,secret456,bob@example.com +``` + +**问题**: ❌ **password字段未被屏蔽,直接导出** + +##### 问题3: 导出文件存在数据泄露风险 + +``` +导出后的文件位置: + - 服务器共享目录 (/shared/exports/) + - 其他用户可能有读权限 + - 文件未加密 + - 没有访问审计 +``` + +**风险**: +- 敏感数据以明文形式存储在共享目录 +- 可被其他有权限的用户访问 +- 可被复制或传播 +- 难以追踪数据流向 + +#### 与下载功能的对比 + +| 维度 | 下载功能 (resultsetToExcel) | 导出功能 (Pipeline) | +|-----|------------------------|------------------| +| **触发方式** | REST API调用 | Pipeline脚本执行 | +| **数据流向** | 服务器 → 客户端浏览器 | 服务器 → 服务器目录 | +| **格式转换** | FsRestfulApi中实现 | Pipeline执行器实现 | +| **敏感字段屏蔽** | ❌ 不支持 | ❌ 不支持 | +| **行数限制** | ✅ 支持 (默认5000) | ❌ 不限制 | +| **权限检查** | ✅ checkIsUsersDirectory() | ⚠️ 仅文件系统级别 | +| **审计日志** | ✅ 有日志记录 | ⚠️ 仅引擎执行日志 | +| **文件访问控制** | ✅ 单次下载后用户控制 | ⚠️ 服务器文件系统权限 | + +#### 完整的Pipeline导出执行链路 + +``` +前端 resultsExport.vue + ↓ [生成Pipeline代码] +from /user/hadoop/linkis/result.dolphin to /shared/export/file.csv + ↓ [提交到Workbench执行] +PipelineEngineConnExecutor.executeLine() + ↓ [正则解析] +sourcePath = /user/hadoop/linkis/result.dolphin +destPath = /shared/export/file.csv + ↓ [选择执行器] +PipelineExecutorSelector.select() → CSVExecutor + ↓ [执行导出] +CSVExecutor.execute() + ├─ FSFactory.getFs(sourcePath) + ├─ FileSource.create(sourcePath, fs) + │ └─ ResultSetFactory.getResultSetByPath() + │ └─ ResultSetReader.getResultSetReader() + │ └─ 读取dolphin文件 (含所有字段) + ├─ CSVFsWriter.getCSVFSWriter() + ├─ fileSource.addParams("nullValue", "NULL") + └─ fileSource.write(csvWriter) + ├─ ResultsetFileSource.shuffle() [仅NULL值处理] + ├─ StorageCSVWriter.addMetaData() [写入所有列名] + └─ StorageCSVWriter.addRecord() [写入所有数据] + ↓ +输出文件: /shared/export/file.csv (包含所有敏感字段) +``` + +**关键发现**: +- 整个链路中**没有任何一个环节**检查或过滤敏感字段 +- 所有字段从dolphin文件原样转换到目标格式 +- 用户可以轻松绕过任何前置的敏感数据检查 + +--- + +## 3. 需求详细说明 + +### 3.1 核心需求 + +**在结果集下载和导出时支持敏感字段屏蔽功能,与查看功能保持一致的安全策略,全面堵塞敏感数据泄露渠道。** + +**涉及的三个功能模块**: +1. ✅ **结果集查看** (`/api/rest_j/v1/filesystem/openFile`) - 已支持屏蔽 +2. ❌ **结果集下载** (`/api/rest_j/v1/filesystem/resultsetToExcel`, `resultsetsToExcel`) - 需要支持 +3. ❌ **结果集导出** (Pipeline引擎: `CSVExecutor`, `ExcelExecutor`) - 需要支持 + +### 3.2 功能要求 + +#### 3.2.1 参数设计 + +##### (1) 下载接口参数扩展 + +在`resultsetToExcel`和`resultsetsToExcel`两个接口中**新增可选参数**: + +| 参数名 | 类型 | 是否必填 | 默认值 | 说明 | +|-------|------|---------|-------|------| +| maskedFieldNames | String | 否 | null | 需要屏蔽的字段名,多个字段用逗号分隔 | + +**示例请求**: + +``` +GET /api/rest_j/v1/filesystem/resultsetToExcel?path=/user/result.dolphin&outputFileType=csv&maskedFieldNames=password,apikey,ssn +``` + +##### (2) Pipeline导出语法扩展 ⭐ + +**职责划分**: +- **其他团队**: 负责前端交互和Pipeline代码生成（包含屏蔽字段） +- **我们团队**: 负责Pipeline引擎执行（解析语法并应用屏蔽逻辑） + +**新增Pipeline语法**: + +``` +from <源路径> to <目标路径> without "<字段名1,字段名2,...>" +``` + +**语法规则**: +- `without` 关键字后跟屏蔽字段列表 +- 字段名用**双引号**包裹 +- 多个字段用**逗号分隔**（不区分大小写） +- 双引号内可包含空格 + +**示例**: + +```sql +-- 示例1: 屏蔽单个字段 +from /user/result.dolphin to /export/file.csv without "password" + +-- 示例2: 屏蔽多个字段 +from /user/result.dolphin to /export/users.xlsx without "password,apikey,credit_card" + +-- 示例3: 字段名包含空格 +from /user/result.dolphin to /export/data.csv without "user password, api key, credit card" + +-- 示例4: 不屏蔽（保持原语法） +from /user/result.dolphin to /export/file.csv +``` + +**语法兼容性**: +- ✅ 向后兼容：不使用`without`子句时，保持原有行为 +- ✅ 大小写不敏感：`WITHOUT`、`without`、`Without`均可 +- ✅ 空格容忍：关键字前后的空格会被自动处理 + +#### 3.2.2 屏蔽规则 + +1. **字段匹配** + - 不区分大小写 + - 精确匹配字段名(columnName) + - 支持多字段,使用逗号分隔 + +2. **屏蔽方式** + - 完全移除屏蔽字段列(而非替换为空值或掩码) + - 同时处理元数据(metadata)和数据内容(fileContent) + - 保持与`openFile`接口的一致性 + +3. **异常处理** + - 如果指定的屏蔽字段不存在,不报错,正常导出 + - 如果所有字段都被屏蔽,返回空结果集(仅包含结果集结构) + +#### 3.2.3 兼容性要求 + +1. **向后兼容**: 不传`maskedFieldNames`参数时,保持原有行为(导出完整数据) +2. **格式兼容**: 支持CSV和XLSX两种导出格式 +3. **性能要求**: 字段屏蔽逻辑不应显著影响导出性能 + +--- + +### 3.3 技术实现要求 + +#### 3.3.1 代码复用 + +- **复用现有方法**: 直接复用`openFile`中已实现的以下方法: + - `filterMaskedFieldsFromMetadata()` (FsRestfulApi.java 行841-858) + - `removeFieldsFromContent()` (FsRestfulApi.java 行787-838) + +- **考虑重构**: 如果方法访问级别不合适,建议将这两个方法: + - 从`private`修改为`protected`或提取到工具类 + - 或直接复制到下载方法和Pipeline执行器中 + +#### 3.3.2 实现位置 + +##### (1) 下载功能实现位置 + +**文件**: `FsRestfulApi.java` + +**修改方法**: +1. `resultsetToExcel()` (行972-1084) +2. `resultsetsToExcel()` (行1105-1189) + +**关键修改点**: + +```java +// 在fileSource.write(fsWriter)之前添加字段过滤逻辑 + +if (StringUtils.isNotBlank(maskedFieldNames)) { + Set maskedFields = + new HashSet<>(Arrays.asList(maskedFieldNames.toLowerCase().split(","))); + + // 获取元数据并过滤 + // 修改FileSource或Writer以支持字段过滤 + // 具体实现需要根据Linkis Storage层的架构决定 +} +``` + +##### (2) Pipeline导出功能实现位置 + +**核心修改**: `PipelineEngineConnExecutor.scala` + +**关键变更**: 扩展正则表达式以支持`without`子句 + +```scala +// 原有正则（仅支持 from ... to ...） +val regex = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s?".r + +// 🆕 新正则（支持 from ... to ... without "..."） +val regexWithMask = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s+without\\s+\"([^\"]+)\"\\s*".r +val regexNormal = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s*".r +``` + +**执行逻辑修改**: + +```scala +// PipelineEngineConnExecutor.scala (行69-89 修改) +override def executeLine( + engineExecutorContext: EngineExecutionContext, + code: String +): ExecuteResponse = { + + thread = Thread.currentThread() + + // 🔍 优先匹配带without子句的语法 + code match { + // 情况1: 包含屏蔽字段 + case regexWithMask(sourcePath, destPath, maskedFields) => + logger.info(s"Pipeline execution with masked fields: $maskedFields") + + // 将屏蔽字段传递给执行器 + val enhancedOptions = new util.HashMap[String, String](newOptions) + enhancedOptions.put("pipeline.masked.field.names", maskedFields) + + PipelineExecutorSelector + .select(sourcePath, destPath, enhancedOptions) + .execute(sourcePath, destPath, engineExecutorContext) + + // 情况2: 不包含屏蔽字段（保持原有行为） + case regexNormal(sourcePath, destPath) => + logger.info(s"Pipeline execution without masking") + + PipelineExecutorSelector + .select(sourcePath, destPath, newOptions) + .execute(sourcePath, destPath, engineExecutorContext) + + // 情况3: 语法错误 + case _ => + throw new PipeLineErrorException( + ILLEGAL_OUT_SCRIPT.getErrorCode, + ILLEGAL_OUT_SCRIPT.getErrorDesc + ". Syntax: from to [without \"fields\"]" + ) + } +} +``` + +**正则表达式详解**: + +```scala +// 正则结构分析 +val regexWithMask = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s+without\\s+\"([^\"]+)\"\\s*".r + +// 组成部分: +// (?i) - 大小写不敏感 +// \\s* - 可选的前导空格 +// from - 关键字 "from" +// \\s+ - 必需的空格 +// (\\S+) - 第1组: 源路径（非空白字符） +// \\s+ - 必需的空格 +// to - 关键字 "to" +// \\s+ - 必需的空格 +// (\\S+) - 第2组: 目标路径（非空白字符） +// \\s+ - 必需的空格 +// without - 关键字 "without" +// \\s+ - 必需的空格 +// \" - 左双引号 +// ([^\"]+) - 第3组: 屏蔽字段列表（除双引号外的任意字符） +// \" - 右双引号 +// \\s* - 可选的尾随空格 +``` + +**测试用例**: + +```scala +// 测试1: 标准语法 +"from /a/b.dolphin to /c/d.csv without \"password,apikey\"" +// 匹配结果: +// sourcePath = "/a/b.dolphin" +// destPath = "/c/d.csv" +// maskedFields = "password,apikey" + +// 测试2: 大小写不敏感 +"FROM /a/b.dolphin TO /c/d.csv WITHOUT \"password\"" +// 匹配成功 + +// 测试3: 字段名包含空格 +"from /a/b.dolphin to /c/d.csv without \"user password, api key\"" +// 匹配结果: +// maskedFields = "user password, api key" + +// 测试4: 兼容原语法 +"from /a/b.dolphin to /c/d.csv" +// 匹配 regexNormal，maskedFields为空 + +// 测试5: 语法错误（缺少引号） +"from /a/b.dolphin to /c/d.csv without password" +// 不匹配任何正则，抛出异常 +``` + +--- + +**修改文件**: `linkis-engineconn-plugins/pipeline/src/main/scala/org/apache/linkis/manager/engineplugin/pipeline/executor/` + +**修改文件清单**: +1. ✅ `PipelineEngineConnExecutor.scala` - 正则解析和参数传递 +2. ✅ `CSVExecutor.scala` - 读取options中的屏蔽字段 +3. ✅ `ExcelExecutor.scala` - 读取options中的屏蔽字段 + +**无需修改**: +- ❌ 前端 `resultsExport.vue` - 由其他团队负责 +- ❌ `PipelineEngineConfiguration.scala` - 参数通过options传递，无需新增配置 + +--- + +#### 3.3.3 实现难点分析 + +**挑战1**: `openFile`与下载/导出接口的数据流处理方式不同 + +- `openFile`: 调用`fileSource.collect()`获取完整数据后过滤 +- `下载/导出接口`: 调用`fileSource.write(fsWriter)`直接流式写入输出流 + +**挑战2**: Pipeline执行器基于Scala实现,需要在Scala代码中实现字段过滤 + +**挑战3**: 需要在多个层级传递屏蔽字段参数 + +``` +前端 resultsExport.vue + ↓ startupMap参数 +Entrance (任务提交) + ↓ JobRequest.params +PipelineEngineConnExecutor + ↓ options传递 +CSVExecutor / ExcelExecutor + ↓ 应用屏蔽逻辑 +FileSource / Writer +``` + +**解决方案选型**: + +##### 已确定方案：方案A - 在执行器中收集数据后过滤 + +**技术选型理由**: +- ✅ 实现简单，可快速交付 +- ✅ 复用FsRestfulApi中已有的字段过滤逻辑 +- ✅ 不需要修改Storage层，风险可控 +- ✅ 适合当前MVP需求 + +**技术限制**: +- ⚠️ 对大结果集（>10万行）有内存压力 +- ⚠️ 性能相对流式方案较低 +- 💡 可通过结果集大小限制规避风险 + +**实现步骤**: + +1. **PipelineEngineConnExecutor修改** (正则解析) + +已在上述"(2) Pipeline导出功能实现位置"中详细说明。 + +2. **CSVExecutor实现** (`CSVExecutor.scala`) + +```scala +override def execute(sourcePath: String, destPath: String, context: EngineExecutionContext): ExecuteResponse = { + // 1. 🆕 从options获取屏蔽字段参数（由PipelineEngineConnExecutor传入） + val maskedFieldNames = options.getOrDefault("pipeline.masked.field.names", "") + + // 2. 验证源文件 + if (!sourcePath.contains(STORAGE_RS_FILE_SUFFIX.getValue)) { + throw new PipeLineErrorException(EXPROTING_MULTIPLE.getErrorCode, EXPROTING_MULTIPLE.getErrorDesc) + } + if (!FileSource.isResultSet(sourcePath)) { + throw new PipeLineErrorException(NOT_A_RESULT_SET_FILE.getErrorCode, NOT_A_RESULT_SET_FILE.getErrorDesc) + } + + // 3. 创建文件系统 + val sourceFsPath = new FsPath(sourcePath) + val destFsPath = new FsPath(destPath) + val sourceFs = FSFactory.getFs(sourceFsPath) + sourceFs.init(null) + val destFs = FSFactory.getFs(destFsPath) + destFs.init(null) + + // 4. 创建FileSource + val fileSource = FileSource.create(sourceFsPath, sourceFs) + if (!FileSource.isTableResultSet(fileSource)) { + throw new PipeLineErrorException(NOT_A_TABLE_RESULT_SET.getErrorCode, NOT_A_TABLE_RESULT_SET.getErrorDesc) + } + + // 5. 获取配置参数 + var nullValue = options.getOrDefault(PIPELINE_OUTPUT_SHUFFLE_NULL_TYPE, "NULL") + if (BLANK.equalsIgnoreCase(nullValue)) nullValue = "" + + // 6. 创建输出流和Writer + val outputStream = destFs.write(destFsPath, PIPELINE_OUTPUT_ISOVERWRITE_SWITCH.getValue(options)) + OutputStreamCache.osCache.put(engineExecutionContext.getJobId.get, outputStream) + + val csvWriter = CSVFsWriter.getCSVFSWriter( + PIPELINE_OUTPUT_CHARSET_STR.getValue(options), + PIPELINE_FIELD_SPLIT_STR.getValue(options), + PIPELINE_FIELD_QUOTE_RETOUCH_ENABLE.getValue(options), + outputStream + ) + + // 7. 🔑 方案A核心逻辑：判断是否需要屏蔽字段 + try { + if (StringUtils.isNotBlank(maskedFieldNames)) { + logger.info(s"Applying field masking: $maskedFieldNames") + + // 7.1 解析屏蔽字段列表 + val maskedFields = maskedFieldNames.toLowerCase().split(",").map(_.trim).toSet + + // 7.2 收集完整数据 + val collectedData = fileSource.collect() + + // 7.3 过滤字段并写入 + filterAndWriteData(collectedData, maskedFields, csvWriter, nullValue) + + } else { + // 原有流式写入逻辑（无屏蔽） + logger.info("No field masking, using stream write") + fileSource.addParams("nullValue", nullValue).write(csvWriter) + } + } finally { + // 8. 资源清理 + IOUtils.closeQuietly(csvWriter) + IOUtils.closeQuietly(fileSource) + IOUtils.closeQuietly(sourceFs) + IOUtils.closeQuietly(destFs) + } + + super.execute(sourcePath, destPath, engineExecutionContext) +} + +// 🆕 字段过滤和写入方法 +private def filterAndWriteData( + collectedData: Array[Pair[Object, ArrayList[String[]]]], + maskedFields: Set[String], + csvWriter: CSVFsWriter, + nullValue: String +): Unit = { + + collectedData.foreach { pair => + // 获取元数据和内容 + val metadata = pair.getFirst.asInstanceOf[Array[util.Map[String, Any]]] + val content = pair.getSecond + + // 计算需要保留的列索引 + val retainedIndices = metadata.zipWithIndex + .filter { case (col, _) => + val columnName = col.get("columnName").toString.toLowerCase() + !maskedFields.contains(columnName) // 不在屏蔽列表中的字段 + } + .map(_._2) + .toList + + logger.info(s"Retained columns: ${retainedIndices.size}/${metadata.length}") + + // 过滤元数据 + val filteredMetadata = retainedIndices.map(i => metadata(i)).toArray + val tableMetaData = new TableMetaData( + filteredMetadata.map { col => + Column( + col.get("columnName").toString, + DataType.toDataType(col.get("dataType").toString), + col.get("comment").toString + ) + } + ) + + // 写入元数据 + csvWriter.addMetaData(tableMetaData) + + // 过滤并写入内容 + content.forEach { row => + val filteredRow = retainedIndices.map { i => + if (i < row.length) { + val value = row(i) + // 处理NULL值 + if (value == null || value.equals("NULL")) nullValue else value + } else { + nullValue + } + }.toArray + + csvWriter.addRecord(new TableRecord(filteredRow)) + } + + csvWriter.flush() + } +} +``` + +3. **ExcelExecutor实现** (`ExcelExecutor.scala`) + - 实现逻辑与CSVExecutor类似 + - 区别在于使用`ExcelFsWriter`替代`CSVFsWriter` + - 支持单结果集和多结果集模式 + +4. **无需修改的部分** + - ❌ `PipelineEngineConfiguration.scala` - 不需要新增配置项 + - ❌ `resultsExport.vue` - 由其他团队负责代码生成 + +--- + +**其他备选方案** (供未来优化参考): + +
+方案B: 扩展Writer实现 (性能优化方案) + +- 创建`MaskedFieldsCSVFsWriter`和`MaskedFieldsExcelFsWriter` +- 在Writer内部实现流式字段过滤 +- 优点: 内存友好，性能优越 +- 缺点: 需修改Storage层，开发周期长 + +
+ +
+方案C: FileSource原生支持 (终极方案) + +- 在`FileSource`中添加`excludeColumns()`方法 +- 架构层面的解决方案，对上层透明 +- 优点: 最优雅，所有场景受益 +- 缺点: 影响范围大，需深度测试 + +
+ +
+方案D: Decorator模式 (方案B的优化版) + +- 不修改现有Writer，遵循开闭原则 +- 使用装饰器包装Writer实现字段过滤 +- 优点: 灵活，可组合 +- 缺点: 增加代码复杂度 + +
+ +**实施建议**: 先实现方案A快速上线，后续根据性能监控考虑升级到方案D + +--- + +### 3.4 配置项说明 + +#### 现有配置项 + +| 配置项 | 默认值 | 说明 | +|-------|-------|------| +| wds.linkis.workspace.resultset.download.is.limit | true | 是否限制下载大小 | +| wds.linkis.workspace.resultset.download.maxsize.csv | 5000 | CSV下载最大行数 | +| wds.linkis.workspace.resultset.download.maxsize.excel | 5000 | Excel下载最大行数 | + +#### 新增配置项（方案A专用） + +| 配置项 | 默认值 | 说明 | 重要性 | +|-------|-------|------|--------| +| **pipeline.masked.field.names** | "" | Pipeline导出时屏蔽的字段名列表（逗号分隔） | 核心功能 | +| **pipeline.export.max.rows** | 100000 | Pipeline导出时允许的最大行数（启用屏蔽时） | ⚠️ 内存保护 | +| **pipeline.export.memory.check.enabled** | true | 是否启用内存检查 | ⚠️ 风险控制 | +| **pipeline.export.memory.threshold** | 0.8 | 内存使用阈值（占总内存比例） | ⚠️ 风险控制 | + +#### 配置建议 + +**生产环境推荐配置**: + +```properties +# 启用结果集大小限制（方案A必需） +pipeline.export.max.rows=50000 + +# 启用内存检查 +pipeline.export.memory.check.enabled=true +pipeline.export.memory.threshold=0.75 + +# 下载功能限制（保持现有） +wds.linkis.workspace.resultset.download.is.limit=true +wds.linkis.workspace.resultset.download.maxsize.csv=5000 +wds.linkis.workspace.resultset.download.maxsize.excel=5000 +``` + +**开发/测试环境配置**: + +```properties +# 可适当放宽限制 +pipeline.export.max.rows=100000 +pipeline.export.memory.check.enabled=false +``` + +**内存充足环境配置**: + +```properties +# 服务器内存>=32GB时可考虑 +pipeline.export.max.rows=500000 +pipeline.export.memory.threshold=0.85 +``` + +--- + +## 4. 风险评估 + +### 4.1 技术风险 + +| 风险 | 等级 | 应对措施 | 备注 | +|-----|------|---------|------| +| **方案A内存溢出风险** | 中 | 1. 配置结果集导出行数上限（建议10万行）
2. 添加内存监控和告警
3. 大结果集提示用户分批导出 | 方案A主要风险 | +| Storage层兼容性问题 | 低 | 充分的兼容性测试，确保Scala/Java互操作正常 | - | +| 性能下降 | 低 | 1. 仅在指定屏蔽字段时启用过滤逻辑
2. 不影响未启用屏蔽的导出性能 | - | +| Pipeline参数传递失败 | 低 | 1. 参数传递链路日志记录
2. 异常情况降级为不屏蔽 | - | +| 字段过滤逻辑错误 | 中 | 1. 完整的单元测试覆盖
2. 与openFile功能对比测试 | 需充分测试 | + +### 4.2 业务风险 + +| 风险 | 等级 | 应对措施 | 备注 | +|-----|------|---------|------| +| 向后兼容性问题 | 低 | 新增可选参数，不传参数时保持原有行为 | - | +| 误屏蔽正常字段 | 低 | 1. 明确文档说明字段名匹配规则
2. 前端提供字段名预览和校验 | - | +| 用户体验影响 | 低 | 1. 前端提供友好的配置界面
2. 屏蔽字段输入支持自动补全 | 可选优化 | +| 大结果集导出超时 | 中 | 1. 方案A会增加导出时间
2. 建议限制行数上限
3. 显示导出进度提示 | 方案A特有 | + +### 4.3 方案A的特殊风险控制 + +#### 风险1：内存溢出 + +**触发条件**: +- 结果集行数 > 10万行 +- 结果集列数 > 1000列 +- 并发导出任务过多 + +**监控指标**: +```scala +// 添加内存使用监控 +val runtime = Runtime.getRuntime +val usedMemory = runtime.totalMemory() - runtime.freeMemory() +if (usedMemory > MAX_MEMORY_THRESHOLD) { + logger.warn(s"Memory usage high: $usedMemory bytes") + throw new PipeLineErrorException("Memory limit exceeded") +} +``` + +**应对措施**: +1. **配置层控制**: 新增配置项限制导出行数 + ```scala + val PIPELINE_EXPORT_MAX_ROWS = CommonVars("pipeline.export.max.rows", 100000) + ``` + +2. **运行时检查**: 在collect()前检查结果集大小 + ```scala + val totalLine = fileSource.getTotalLine + if (totalLine > PIPELINE_EXPORT_MAX_ROWS.getValue) { + throw new PipeLineErrorException( + s"Result set too large: $totalLine rows, max allowed: ${PIPELINE_EXPORT_MAX_ROWS.getValue}" + ) + } + ``` + +3. **用户提示**: 前端显示结果集大小，超过阈值时警告 + +#### 风险2：性能下降 + +**影响评估**: +- 原流式写入: 无需加载全部数据到内存 +- 方案A: 需先collect()全部数据，再过滤，再写入 +- **预估性能损失**: 30-50% (取决于结果集大小) + +**缓解措施**: +1. 仅在指定屏蔽字段时启用collect模式 +2. 未指定屏蔽字段时保持原流式写入 +3. 添加性能日志，监控导出耗时 + +--- + +## 5. 变更历史 + +| 版本 | 日期 | 变更内容 | 作者 | +|-----|------|---------|------| +| v1.0 | 2025-10-27 | 初始版本 - 完成需求分析和技术方案设计 | Claude Code | + +--- + +**文档结束** + diff --git a/linkis-commons/linkis-scheduler/src/test/java/org/apache/linkis/scheduler/queue/PriorityLoopArrayQueueTest.java b/linkis-commons/linkis-scheduler/src/test/java/org/apache/linkis/scheduler/queue/PriorityLoopArrayQueueTest.java deleted file mode 100644 index cd45a991f1..0000000000 --- a/linkis-commons/linkis-scheduler/src/test/java/org/apache/linkis/scheduler/queue/PriorityLoopArrayQueueTest.java +++ /dev/null @@ -1,234 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.linkis.scheduler.queue; - -import org.apache.linkis.scheduler.queue.fifoqueue.FIFOGroup; - -import java.util.Random; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicInteger; - -import scala.Option; - -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -class PriorityLoopArrayQueueTest { - AtomicInteger productCounter = new AtomicInteger(); - AtomicInteger consumerCounter = new AtomicInteger(); - - @Test - public void testConcurrentPutAndTake() throws Exception { - AtomicInteger counter = new AtomicInteger(); - FIFOGroup group = new FIFOGroup("test", 5000, 5000); - PriorityLoopArrayQueue queue = new PriorityLoopArrayQueue(group); - boolean testFlag = false; - - if (testFlag) { - // 获取开始时间的毫秒数 - long startTime = System.currentTimeMillis(); - // 10s的毫秒数 - long threeMinutesInMillis = 10 * 1000; - int genLen = 50; - int getLen = 70; - final CountDownLatch latch = new CountDownLatch(genLen + getLen + 1); - // 5 个生产者 - for (int i = 0; i < genLen; i++) { - final int id = i; - new Thread( - () -> { - try { - Thread.sleep(100 * id); - latch.countDown(); - latch.await(); - } catch (InterruptedException e) { - e.printStackTrace(); - } - System.out.println(Thread.currentThread().getName() + "开始生产："); - while ((System.currentTimeMillis() - startTime) < threeMinutesInMillis) { - // 生产 - try { - Thread.sleep(getRandom(200)); - product(counter, queue); - product(counter, queue); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - // 消费 - // consume(queue); - } - System.out.println(Thread.currentThread().getName() + "结束生产："); - }, - "生产t-" + i) - .start(); - } - // 5 个消费者 - for (int i = 0; i < getLen; i++) { - final int id = i; - new Thread( - () -> { - try { - Thread.sleep(getRandom(200)); - latch.countDown(); - latch.await(); - } catch (InterruptedException e) { - e.printStackTrace(); - } - System.out.println(Thread.currentThread().getName() + "开始消费："); - while (true) { - try { - Thread.sleep(getRandom(200)); - // 消费 - consume(queue); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - }, - "消费t-" + i) - .start(); - } - new Thread( - () -> { - try { - Thread.sleep(100); - latch.countDown(); - latch.await(); - } catch (InterruptedException e) { - e.printStackTrace(); - } - System.out.println(Thread.currentThread().getName() + "开始获取当前队列元素："); - while ((System.currentTimeMillis() - startTime) < threeMinutesInMillis * 2) { - try { - Thread.sleep(5000); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - System.out.println("生产大小：" + productCounter.get()); - System.out.println("消费大小：" + consumerCounter.get()); - System.out.println("队列当前大小：" + queue.size()); - // 需要 去掉私有测试 - // System.out.println("index size: " + queue.indexMap().size()); - // System.out.println("cache size: " + queue.fixedSizeCollection().size()); - } - }) - .start(); - Thread.sleep(threeMinutesInMillis * 2); - System.out.println( - "product:" + productCounter.get() + ", consumer: " + consumerCounter.get()); - // 需要 去掉私有测试 - // Assertions.assertEquals(1000, queue.fixedSizeCollection().size()); - Assertions.assertEquals(productCounter.get(), consumerCounter.get()); - } - } - - // 消费 - private void consume(PriorityLoopArrayQueue queue) { - SchedulerEvent take = null; - try { - take = queue.take(); - consumerCounter.addAndGet(1); - } catch (Exception e) { - throw new RuntimeException(e); - } - printEvent("消费", take); - } - - // 生产 - private void product(AtomicInteger counter, PriorityLoopArrayQueue queue) { - int i1 = counter.addAndGet(1); - // 1000-重要，100-普通，10-不重要 - int[] proArr = {1000, 100, 10}; - int priority = getRandom(3); - String name = "item-" + i1 + "-" + priority; - System.out.println("生产：" + name); - Option offer = queue.offer(getJob(name, proArr[priority])); - if (offer.nonEmpty()) { - productCounter.addAndGet(1); - Option schedulerEventOption = queue.get((int) offer.get()); - printEvent("get：", schedulerEventOption.get()); - } else { - System.out.println("当前队列已满，大小：" + queue.size()); - } - } - - @Test - void testFinally() {} - - @Test - void enqueue() { - // 压测 offer take get - FIFOGroup group = new FIFOGroup("test", 100, 100); - PriorityLoopArrayQueue queue = new PriorityLoopArrayQueue(group); - Option idx = queue.offer(getJob("job1-1", 1)); - // 插入测试 - Assertions.assertEquals(1, (int) idx.get()); - queue.offer(getJob("job2", 2)); - queue.offer(getJob("job3", 3)); - queue.offer(getJob("job1-2", 1)); - queue.offer(getJob("job5", 5)); - queue.offer(getJob("item1-3", 1)); - queue.offer(getJob("item6-1", 6)); - queue.offer(getJob("item4", 4)); - queue.offer(getJob("item6-2", 6)); - // peek 测试 - Option peek = queue.peek(); - Assertions.assertEquals("item6-1", peek.get().getId()); - while (queue.size() > 1) { - queue.take(); - } - SchedulerEvent event = queue.take(); - // 优先级，以及先进先出测试 - Assertions.assertEquals("item1-3", event.getId()); - Assertions.assertEquals(1, event.priority()); - Assertions.assertEquals(6, event.getIndex()); - // 缓存测试，需要设置 linkis.fifo.priority.queue.max.cache.size 为 5 - // Assertions.assertThrows( - // IllegalArgumentException.class, - // () -> { - // queue.get(7); - // }); - } - - private void printEvent(String opt, SchedulerEvent event) { - System.out.println( - "【" - + Thread.currentThread().getName() - + "】" - + opt - + ":" - + event.getId() - + ", priority: " - + event.getPriority() - + ", index: " - + event.getIndex()); - } - - private int getRandom(int bound) { - Random rand = new Random(); - int res = rand.nextInt(bound); - return res; - } - - private UserJob getJob(String name, int priority) { - UserJob job = new UserJob(); - job.setId(name); - job.setPriority(priority); - return job; - } -} diff --git a/linkis-public-enhancements/linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java b/linkis-public-enhancements/linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java index d70d2272ae..60143cafaf 100644 --- a/linkis-public-enhancements/linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java +++ b/linkis-public-enhancements/linkis-pes-publicservice/src/main/java/org/apache/linkis/filesystem/restful/api/FsRestfulApi.java @@ -648,6 +648,9 @@ public Message openFile( throw WorkspaceExceptionManager.createException(80036, path); } + // 检查是否为管理台请求（enableLimit=true） + boolean enableLimitResult = Boolean.parseBoolean(enableLimit); + String userName = ModuleUserUtils.getOperationUser(req, "openFile " + path); LoggerUtils.setJobIdMDC("openFileThread_" + userName); LOGGER.info("userName {} start to open File {}", userName, path); @@ -695,7 +698,7 @@ public Message openFile( 80034, FILESYSTEM_RESULTSET_ROW_LIMIT.getValue()); } - if (StringUtils.isNotBlank(enableLimit) && "true".equals(enableLimit)) { + if (enableLimitResult) { LOGGER.info("set enable limit for thread: {}", Thread.currentThread().getName()); LinkisStorageConf.enableLimitThreadLocal().set(enableLimit); // 组装列索引 @@ -769,7 +772,8 @@ public Message openFile( ResultUtils.removeFieldsFromContent(resultmap, filteredContent, maskedFields); } // 优先截取大字段 - if (LinkisStorageConf.FIELD_TRUNCATION_ENABLED()) { + if (LinkisStorageConf.FIELD_TRUNCATION_ENABLED() && enableLimitResult) { + // 管理台请求(enableLimit=true)不进行字段长度拦截，兼容旧逻辑 FieldTruncationResult fieldTruncationResult = ResultUtils.detectAndHandle( filteredMetadata, @@ -779,7 +783,9 @@ public Message openFile( if (fieldTruncationResult.isHasOversizedFields()) { // 检测到超长字段 if (null == truncateColumn) { + message.data("type", fileSource.getFileSplits()[0].type()); message.data("oversizedFields", fieldTruncationResult.getOversizedFields()); + message.data("display_prohibited", true); message.data("zh_msg", truncateColumn_msg); message.data("en_msg", truncateColumn_en_msg); return message; @@ -806,7 +812,7 @@ public Message openFile( } } if (StringUtils.isNotBlank(maskedFieldNames) - || LinkisStorageConf.FIELD_TRUNCATION_ENABLED()) { + || (LinkisStorageConf.FIELD_TRUNCATION_ENABLED() && enableLimitResult)) { message.data("metadata", filteredMetadata).data("fileContent", filteredContent); } else { // 不执行字段屏蔽也不执行字段截取