Skip to content

Commit fe90109

Browse files
millerjpmillerjp
committed
writetime
1 parent 3231ae2 commit fe90109

File tree

7 files changed

+1512
-10
lines changed

7 files changed

+1512
-10
lines changed

‎libs/async-cassandra-bulk/IMPLEMENTATION_NOTES.md‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
# Implementation Notes - Writetime Export Feature
2+
3+
## Session Context
4+
This implementation was completed across multiple sessions due to context limits. Here's what was accomplished:
5+
6+
### Session 1 (Previous)
7+
- Initial TDD setup and unit tests
8+
- Basic writetime implementation
9+
- Initial integration tests
10+
11+
### Session 2 (Current)
12+
- Fixed all test failures
13+
- Enhanced checkpoint/resume functionality
14+
- Added comprehensive integration tests
15+
- Fixed all linting errors
16+
17+
## Key Technical Decisions
18+
19+
### 1. Query Generation Strategy
20+
We modify the CQL query to include WRITETIME() functions:
21+
```sql
22+
-- Original
23+
SELECT id, name, value FROM table
24+
25+
-- With writetime
26+
SELECT id, name, WRITETIME(name) AS name_writetime, value, WRITETIME(value) AS value_writetime FROM table
27+
```
28+
29+
### 2. Counter Column Handling
30+
Counter columns don't support WRITETIME() in Cassandra, so we:
31+
1. Detect counter columns via `col_meta.cql_type == 'counter'`
32+
2. Exclude them from writetime query generation
33+
3. Exclude them from CSV/JSON headers
34+
35+
### 3. Checkpoint Enhancement
36+
The checkpoint now includes the full export configuration:
37+
```python
38+
checkpoint = {
39+
"version": "1.0",
40+
"completed_ranges": [...],
41+
"total_rows": 12345,
42+
"export_config": {
43+
"table": "keyspace.table",
44+
"columns": ["col1", "col2"],
45+
"writetime_columns": ["col1"], # Preserved!
46+
"batch_size": 1000,
47+
"concurrency": 4
48+
}
49+
}
50+
```
51+
52+
### 4. Collection Column Handling
53+
Collection columns (list, set, map) return a list of writetime values:
54+
```python
55+
# Handle list values in WritetimeSerializer
56+
if isinstance(value, list):
57+
if value:
58+
value = value[0] # Use first writetime
59+
else:
60+
return None
61+
```
62+
63+
## Testing Philosophy
64+
65+
All tests follow CLAUDE.md requirements:
66+
1. Test-first development (TDD)
67+
2. Comprehensive documentation in each test
68+
3. Real Cassandra for integration tests (no mocks)
69+
4. Edge cases and error scenarios covered
70+
5. Performance and stress testing included
71+
72+
## Error Handling Evolution
73+
74+
### Initial Issues
75+
1. **TypeError with collections** - Fixed by handling list values
76+
2. **RuntimeError on resume** - Fixed header management
77+
3. **Counter columns** - Fixed by proper type detection
78+
79+
### Resolution Pattern
80+
Each fix followed this pattern:
81+
1. Reproduce in test
82+
2. Understand root cause
83+
3. Implement minimal fix
84+
4. Verify all tests pass
85+
5. Add regression test
86+
87+
## Performance Considerations
88+
89+
1. **Minimal overhead when disabled** - No WRITETIME() in query
90+
2. **Linear scaling** - Overhead proportional to writetime columns
91+
3. **Memory efficient** - Streaming not affected
92+
4. **Checkpoint overhead minimal** - Only adds config to existing checkpoint
93+
94+
## Code Quality
95+
96+
### Linting Compliance
97+
- All F841 (unused variables) fixed
98+
- E722 (bare except) fixed
99+
- F821 (undefined names) fixed
100+
- Import ordering fixed by isort
101+
- Black formatting applied
102+
- Type hints maintained
103+
104+
### Test Coverage
105+
- Unit tests: Query generation, serialization, configuration
106+
- Integration tests: Full export scenarios, error cases
107+
- Stress tests: High concurrency, large datasets
108+
- Example code: Demonstrates all features
109+
110+
## Lessons Learned
111+
112+
1. **Collection columns are tricky** - Always test with maps, lists, sets
113+
2. **Counter columns are special** - Must be detected and excluded
114+
3. **Resume must preserve config** - Users expect same behavior
115+
4. **Token wraparound matters** - Edge cases at MIN/MAX tokens
116+
5. **Real tests find real bugs** - Mocks would have missed several issues
117+
118+
## Future Considerations
119+
120+
1. **Writetime filtering** - Export only recently updated rows
121+
2. **TTL support** - Export TTL alongside writetime
122+
3. **Incremental exports** - Use writetime for change detection
123+
4. **Writetime statistics** - Min/max/avg in export summary
124+
125+
## Maintenance Notes
126+
127+
When modifying this feature:
128+
1. Run full test suite including stress tests
129+
2. Test with real Cassandra cluster
130+
3. Verify checkpoint compatibility
131+
4. Check performance impact
132+
5. Update examples if API changes

‎libs/async-cassandra-bulk/WRITETIME_FILTERING_IMPLEMENTATION.md‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
# Writetime Filtering Implementation - Progress Report
2+
3+
## Overview
4+
Successfully implemented writetime filtering functionality for the async-cassandra-bulk library, allowing users to export rows based on when they were last written to Cassandra.
5+
6+
## Key Features Implemented
7+
8+
### 1. Writetime Filtering Options
9+
- **writetime_after**: Export only rows where ANY/ALL columns were written after a specified timestamp
10+
- **writetime_before**: Export only rows where ANY/ALL columns were written before a specified timestamp
11+
- **writetime_filter_mode**: Choose between "any" (default) or "all" mode for filtering logic
12+
- **Flexible timestamp formats**: Supports ISO strings, unix timestamps (seconds/milliseconds), and datetime objects
13+
14+
### 2. Row-Level Filtering
15+
- Filters entire rows based on writetime values, not individual cells
16+
- ANY mode: Include row if ANY writetime column matches the filter criteria
17+
- ALL mode: Include row only if ALL writetime columns match the filter criteria
18+
- Handles collection columns that return lists of writetime values
19+
20+
### 3. Validation and Safety
21+
- Validates that tables have columns supporting writetime (excludes primary keys and counters)
22+
- Prevents logical errors (e.g., before < after)
23+
- Clear error messages for invalid configurations
24+
- Preserves filter configuration in checkpoints for resume functionality
25+
26+
## Implementation Details
27+
28+
### Files Modified
29+
1. **src/async_cassandra_bulk/operators/bulk_operator.py**
30+
- Added `_parse_writetime_filters()` method for parsing timestamp options
31+
- Added `_parse_timestamp_to_micros()` method for flexible timestamp conversion
32+
- Added `_validate_writetime_options()` method for validation
33+
- Enhanced `export()` method to pass filter parameters to ParallelExporter
34+
35+
2. **src/async_cassandra_bulk/parallel_export.py**
36+
- Added writetime filter parameters to constructor
37+
- Implemented `_should_filter_row()` method for row-level filtering logic
38+
- Enhanced `_export_range()` to apply filtering during export
39+
- Added validation in `export()` to check table has writable columns
40+
- Updated checkpoint functionality to preserve filter configuration
41+
42+
### Files Created
43+
1. **tests/unit/test_writetime_filtering.py**
44+
- Comprehensive unit tests for timestamp parsing
45+
- Tests for various timestamp formats
46+
- Validation logic tests
47+
- Error handling tests
48+
49+
2. **tests/integration/test_writetime_filtering_integration.py**
50+
- Integration tests with real Cassandra 5
51+
- Tests for after/before/range filtering
52+
- Performance comparison tests
53+
- Checkpoint/resume with filtering tests
54+
- Edge case handling tests
55+
56+
## Testing Summary
57+
58+
### Unit Tests (7 tests)
59+
- ✅ test_writetime_filter_parsing - Various timestamp format parsing
60+
- ✅ test_invalid_writetime_filter_formats - Error handling for invalid formats
61+
- ✅ test_export_with_writetime_after_filter - Filter passed to exporter
62+
- ✅ test_export_with_writetime_before_filter - Before filter functionality
63+
- ✅ test_export_with_writetime_range_filter - Both filters combined
64+
- ✅ test_writetime_filter_with_no_writetime_columns - Validation logic
65+
66+
### Integration Tests (7 tests)
67+
- ✅ test_export_with_writetime_after_filter - Real data filtering after timestamp
68+
- ✅ test_export_with_writetime_before_filter - Real data filtering before timestamp
69+
- ✅ test_export_with_writetime_range_filter - Time window filtering
70+
- ✅ test_writetime_filter_with_no_matching_data - Empty result handling
71+
- ✅ test_writetime_filter_performance - Performance impact measurement
72+
- ✅ test_writetime_filter_with_checkpoint_resume - Resume maintains filters
73+
74+
## Usage Examples
75+
76+
### Export Recent Data (Incremental Export)
77+
```python
78+
await operator.export(
79+
table="myks.events",
80+
output_path="recent_events.csv",
81+
format="csv",
82+
options={
83+
"writetime_after": "2024-01-01T00:00:00Z",
84+
"writetime_columns": ["status", "updated_at"]
85+
}
86+
)
87+
```
88+
89+
### Archive Old Data
90+
```python
91+
await operator.export(
92+
table="myks.events",
93+
output_path="archive_2023.json",
94+
format="json",
95+
options={
96+
"writetime_before": "2024-01-01T00:00:00Z",
97+
"writetime_columns": ["*"], # All non-key columns
98+
"writetime_filter_mode": "all" # ALL columns must be old
99+
}
100+
)
101+
```
102+
103+
### Export Specific Time Range
104+
```python
105+
await operator.export(
106+
table="myks.events",
107+
output_path="q2_2024.csv",
108+
format="csv",
109+
options={
110+
"writetime_after": datetime(2024, 4, 1, tzinfo=timezone.utc),
111+
"writetime_before": datetime(2024, 6, 30, 23, 59, 59, tzinfo=timezone.utc),
112+
"writetime_columns": ["event_type", "status", "value"]
113+
}
114+
)
115+
```
116+
117+
## Technical Decisions
118+
119+
1. **Row-Level Filtering**: Chose to filter entire rows rather than individual cells since we're exporting rows, not cells
120+
2. **Microsecond Precision**: Cassandra uses microseconds since epoch for writetime, so all timestamps are converted to microseconds
121+
3. **Flexible Input Formats**: Support multiple timestamp formats for user convenience
122+
4. **ANY/ALL Modes**: Provide flexibility in how multiple writetime values are evaluated
123+
5. **Validation**: Prevent exports on tables that don't support writetime (only PKs/counters)
124+
125+
## Issues Resolved
126+
127+
1. **Test Framework Compatibility**: Converted unittest.TestCase to pytest style
128+
2. **Timestamp Calculations**: Fixed date arithmetic errors in test data
129+
3. **JSON Serialization**: Handled writetime values properly in JSON output
130+
4. **Linting Compliance**: Fixed all 47 linting errors (42 auto-fixed, 5 manual)
131+
132+
## Next Steps
133+
134+
1. Implement TTL export functionality
135+
2. Create combined writetime + TTL tests
136+
3. Update example applications to demonstrate new features
137+
4. Update main documentation
138+
139+
## Commit Summary
140+
141+
Added writetime filtering support to async-cassandra-bulk:
142+
- Filter exports by row writetime (before/after timestamps)
143+
- Support ANY/ALL filtering modes for multiple columns
144+
- Flexible timestamp format parsing
145+
- Comprehensive unit and integration tests
146+
- Full checkpoint/resume support

0 commit comments

Comments
 (0)