-
[OPIK-3056] [BE] Implement CSV generation with S3 multipart upload streaming (#4732)
frostbyte_neo 发布于
2026-01-14 16:39:18 +00:00 -
OPIK-3056: Add database schema and API models for dataset export jobs
-
Revision 2: Address PR comments - fix migration number, author, and field sizes
-
Revision 3: Change file_path to TEXT for unlimited length
-
Revision 4: Add ON UPDATE CURRENT_TIMESTAMP to last_updated_at
-
Revision 5: Add last_updated_by field for audit trail consistency
-
Revision 9: Rename migration file from 000042 to 000043 to avoid conflict
-
OPIK-3056: Add streaming upload method to FileService
-
[OPIK-3056] [BE] Add dataset export job management with Redis stream integration
-
Remove files
-
Revision 2: Address PR comments - add workspace security, fix listener error handling, extract constants, rename polling
-
Fix config issues
-
Revision 5: Remove manual last_updated_at handling (now automatic via DB)
-
Revision 6: Rename DatasetExportJobListener to DatasetExportJobSubscriber and add toggle
-
Revision 7: Split update methods into updateToCompleted and updateToFailed
-
Revision 8: Add unit tests for DatasetExportJobSubscriber lifecycle gating
-
Revision 8: Add last_updated_by field for audit trail
-
Revision 4: Address PR comments - add workspace security, feature toggle, and tests
- Add workspace_id filtering to DatasetExportJobDAO.deleteByIds to prevent cross-tenant deletion
- Add isEnabled() check in CsvDatasetExportService.startExport to prevent jobs when export is disabled
- Add test for disabled export feature
- Update existing tests to mock isEnabled() returning true
- Verify last_updated_at is handled by DB default (ON UPDATE CURRENT_TIMESTAMP)
- Existing DatasetExportJobSubscriberTest already covers lifecycle gating
- Revision 5: Fix DatasetExportJobDAO injection issue
- Remove direct DAO injection from DatasetExportJobService
- Update test to match new constructor signature
- DAO is correctly attached via handle.attach() in transactions
- Fixes Guice injection error on application startup
- All tests pass successfully
- Revision 6: Refactor duplicate NotFoundException tests into parameterized test
- Merge updateJobToCompleted and updateJobToFailed NotFoundException tests
- Use @MethodSource instead of hardcoded test data
- Generate dynamic file paths and error messages
- Reduces test duplication while maintaining coverage
- [OPIK-3056] Implement CSV generation and export processing
- Add CsvDatasetExportProcessor for generating CSV from dataset items
- Implement DatasetExportJobSubscriber to consume export jobs from Redis
- Add DatasetItemDAO.getColumns() method for column discovery
- Update DatasetExportJobService to use TransactionTemplate correctly
- Add comprehensive unit tests for CSV processor and subscriber
- All tests passing (7/7)
-
Fix toggle
-
OPIK-3056: Implement CSV generation with S3 multipart upload streaming
-
Revision 2: Add error message truncation utility and infrastructure updates
-
Fix config
-
Revision 3: Address all PR review comments
- Fix empty dataset handling: Add defaultIfEmpty to getColumns for header-only CSV
- Fix NPE: Handle null error messages in DatasetExportJobSubscriber
- Fix data loss: Refactor buffer loop to process all rows without early return
- Fix exception type: Replace RuntimeException with IllegalStateException
- Fix pagination: Implement proper cursor-based pagination in streamAllItems
- Implement abort: Add abortMultipartUpload to FileService for cleanup
- Fix test config: Enable datasetExport feature in config-test.yml
- Add S3 validation: Clamp minPartSize to S3 minimum (5MB)
- Revision 4: Use StringUtils.defaultIfBlank for NPE handling
Replace manual null check with Apache Commons StringUtils.defaultIfBlank utility method,
following project conventions for handling null strings.- Revision 5: Fix CsvDatasetExportProcessorImplTest for multipart upload
- Update all test verifications to check multipart upload methods (createMultipartUpload,
uploadPart, completeMultipartUpload) instead of the old single upload method - Handle special case for empty datasets which abort multipart and fall back to regular upload
- All 5 tests now pass successfully
- Revision 6: Address new PR review comments
-
Validate maxPartSize against minPartSize to prevent S3 EntityTooSmall errors
- Clamp maxPartSize to Math.max(configuredMaxPartSize, minPartSize)
- Log warning when clamping occurs
- Ensures non-final parts always meet S3's 5MB minimum requirement
-
Extract duplicate NotFoundException check into verifyJobExistsOrThrow helper
- Reduces code duplication in updateJobToCompleted and updateJobToFailed
- Centralizes job existence validation logic
- Improves maintainability
- Revision 7: Simplify minPartSize/maxPartSize validation
Remove intermediate variables and warning logs since S3 has a hard limit of 5MB.
The Math.max() clamping is sufficient - no need to track original config values.- Revision 8: Extract duplicate CSV writing plumbing into helper method
Extract repeated ByteArrayOutputStream + OutputStreamWriter + CSVPrinter setup
into a reusable writeCsv() helper method with ThrowingConsumer interface.Benefits:
- Removes code duplication between createCsvHeader and convertItemToCsvRow
- Centralizes CSV writer creation, flushing, and exception handling
- Simplifies both methods to focus on record data logic only
- Maintains same charset (UTF-8) and CSVFormat.DEFAULT behavior
All tests pass - behavior unchanged.
- Revision 6: Address PR comments on config validation and performance
- Updated config.yml comments to clarify runtime clamping behavior
- Removed unnecessary String allocations in convertItemToCsvRow
- Changed convertItemToCsvRow to return byte[] instead of String
- Updated appendToBuffer to work with byte[] directly
- Preserve database column order (removed sorting)
- Updated test expectations to match database column order
-
OPIK-3056: Address PR comments - user-friendly error message and deterministic column ordering
-
OPIK-3056: Use LinkedHashMap to preserve column insertion order instead of alphabetical sorting
-
OPIK-3056: Fix DatasetItemBatch constructor call with 4th parameter
下载附件
-