apache / apache/paimon

[Feature] PyPaimon Real RowKind Support

Open
#6,735 0 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Motivation

## I. Overview

### 1.1 Problem Description

The PyPaimon `KeyValueDataWriter` was hardcoding all rows' `_VALUE_KIND` to 0 (INSERT), losing information about different row operation types. This prevented proper handling of CDC (Change Data Capture) scenarios with UPDATE and DELETE operations.

**Example Scenario:**
- Write operation 1: User 'Alice' with RowKind=INSERT
- Write operation 2: User 'Alice' updated to 'Alice_updated' with RowKind=UPDATE_AFTER
- Write operation 3: User 'Alice' deleted with RowKind=DELETE
- **Problem**: All three records stored with _VALUE_KIND=0 (INSERT)
- **Expected**: Each record should have correct _VALUE_KIND (0, 2, 3)

### 1.2 Root Cause Analysis

**Write-side Issue**: The `_add_system_fields()` method in `KeyValueDataWriter` hardcoded `_VALUE_KIND` values:

```python
# Before: Always INSERT
value_kind_column = pa.array([0] * num_rows, type=pa.int32())
```

**Impact Areas:**
- **CDC scenarios**: Cannot distinguish UPDATE_BEFORE/UPDATE_AFTER operations
- **Delete handling**: Cannot mark DELETE rows (RowKind=3)
- **Read-side consistency**: Read-side `DropDeleteReader` depends on correct RowKind values
- **Data semantics**: Loss of operational metadata

### 1.3 RowKind Definition

PyPaimon defines four RowKind types:
- **INSERT (0)**: +I - New record insertion
- **UPDATE_BEFORE (1)**: -U - Previous value before update
- **UPDATE_AFTER (2)**: +U - New value after update
- **DELETE (3)**: -D - Record deletion

## II. Technical Solution Design

### 2.1 Core Design Approach

Implement real RowKind extraction from input data by:
1. Reading optional `__row_kind__` column from input RecordBatch
2. Validating RowKind values (0-3 range)
3. Using extracted values for `_VALUE_KIND` field
4. Maintaining backward compatibility (defaulting to INSERT)

### 2.2 Solution Advantages

1. **Full CDC Support**: Correctly handles all row operation types
2. **Backward Compatible**: No breaking changes to existing API
3. **Type-Safe**: Validates data types and value ranges
4. **Well-Tested**: Comprehensive unit test coverage (12 tests)
5. **Production-Ready**: Clean error messages and logging

### 2.3 API Design

**User API - Method 1: Using `__row_kind__` column**
```python
import pyarrow as pa
from pypaimon.table.row.row_kind import RowKind

data = pa.Table.from_pydict({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'__row_kind__': pa.array([
RowKind.INSERT.value, # 0
RowKind.UPDATE_AFTER.value, # 2
RowKind.DELETE.value # 3
], type=pa.int32())
})
write.write_arrow(data)
```

**User API - Method 2: Using row_kinds parameter**
```python
data = pa.Table.from_pydict({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie']
})
write.write_arrow(data, row_kinds=[0, 2, 3])
```

## III. Implementation Details

### 3.1 KeyValueDataWriter Modifications

**File**: `paimon-python/pypaimon/write/writer/key_value_data_writer.py`

#### New Method: `_extract_row_kind_column()`
```python
def _extract_row_kind_column(self, data: pa.RecordBatch, num_rows: int) -> pa.Array:
"""Extract or generate RowKind column from input data.

- Validates data type (must be int32)
- Validates values are in range [0-3]
- Returns extracted column or default INSERT (0) for all rows
"""
```

**Key Features:**
- Checks for `__row_kind__` column presence
- Type validation: `int32` only
- Value validation: 0-3 range check
- Default behavior: All INSERT when column missing
- Logging for debugging

#### Updated Method: `_add_system_fields()`
- Calls `_extract_row_kind_column()` instead of hardcoding
- Removes temporary `__row_kind__` column after extraction
- Maintains backward compatibility

#### New Method: `_deduplicate_by_primary_key()`
- Deduplicates data by primary key
- Preserves latest record (maximum sequence number)
- O(n) time complexity

### 3.2 BatchTableWrite API Enhancement

**File**: `paimon-python/pypaimon/write/batch_table_write.py`

#### Enhanced Methods:
- `write_arrow(table, row_kinds: Optional[List[int]])` - Table-level write with RowKind
- `write_arrow_batch(batch, row_kinds: Optional[List[int]])` - Batch-level write with RowKind

#### New Helper Methods:
- `_add_row_kind_column()` - Adds RowKind column to Table
- `_add_row_kind_to_batch()` - Adds RowKind column to RecordBatch

#### Improved Validation:
- `_validate_pyarrow_schema()` - Now ignores temporary `__row_kind__` column

## IV. Before and After Comparison

### 4.1 Write Data Flow

| Aspect | Before | After |
|--------|--------|-------|
| **RowKind Support** | Hardcoded INSERT (0) | Real RowKind (0-3) |
| **CDC Scenarios** | Not supported | Fully supported |
| **UPDATE Operations** | Lost metadata | Preserved as BEFORE/AFTER |
| **DELETE Operations** | Indistinguishable | Marked as DELETE (3) |
| **API** | No row_kinds parameter | Optional row_kinds parameter |
| **Backward Compatibility** | N/A | 100% compatible |
| **Default Behavior** | All INSERT | INSERT when no RowKind provided |

### 4.2 Example Data Transformation

**Input Data (with RowKind):**
```
id=1, name='Alice', __row_kind__=0 (INSERT)
id=2, name='Bob', __row_kind__=2 (UPDATE_AFTER)
id=3, name='Charlie', __row_kind__=3 (DELETE)
```

**Before Implementation:**
```
_KEY_id=1, name='Alice', _VALUE_KIND=0, _SEQUENCE_NUMBER=1
_KEY_id=2, name='Bob', _VALUE_KIND=0, _SEQUENCE_NUMBER=2 ❌ Should be 2
_KEY_id=3, name='Charlie', _VALUE_KIND=0, _SEQUENCE_NUMBER=3 ❌ Should be 3
```

**After Implementation:**
```
_KEY_id=1, name='Alice', _VALUE_KIND=0, _SEQUENCE_NUMBER=1 ✅ Correct
_KEY_id=2, name='Bob', _VALUE_KIND=2, _SEQUENCE_NUMBER=2 ✅ Correct
_KEY_id=3, name='Charlie', _VALUE_KIND=3, _SEQUENCE_NUMBER=3 ✅ Correct
```

### 4.3 Read-side Benefits

**Without Real RowKind:**
- DropDeleteReader cannot correctly identify DELETE rows
- All rows appear as insertions
- CDC semantics lost during reads

**With Real RowKind:**
- DropDeleteReader correctly filters DELETE rows (RowKind=3)
- UPDATE operations maintain semantic correctness
- Full CDC consistency across write-read pipeline

## V. Performance Analysis

### 5.1 Complexity Analysis

| Operation | Time | Space | Notes |
|-----------|------|-------|-------|
| Extract RowKind | O(n) | O(1) | Single pass validation |
| Deduplication | O(n) | O(m) | m = distinct primary keys |
| Add RowKind column | O(1) | O(n) | PyArrow native operation |
| Schema validation | O(1) | O(1) | Column presence check |

### 5.2 Memory Impact

- Per-row overhead: 4 bytes (int32 RowKind)
- Temporary column: Automatically cleaned up after extraction
- Hash map for dedup: O(m) where m = unique primary keys

## Appendix: Code Examples

### Example 1: Basic RowKind Usage
```python
import pyarrow as pa
from pypaimon.table.row.row_kind import RowKind

# Create data with RowKind information
data = pa.Table.from_pydict({
'user_id': [1, 2, 3, 4],
'name': ['Alice', 'Bob', 'Charlie', 'Dave'],
'__row_kind__': pa.array([
RowKind.INSERT.value, # 0 - New user
RowKind.UPDATE_AFTER.value, # 2 - Updated name
RowKind.DELETE.value, # 3 - User deleted
RowKind.INSERT.value # 0 - New user
], type=pa.int32())
})

# Write with real RowKind values
write.write_arrow(data)
```

### Example 2: Using row_kinds Parameter
```python
# Data without __row_kind__ column
data = pa.Table.from_pydict({
'user_id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie']
})

# Specify RowKind separately
write.write_arrow(data, row_kinds=[0, 2, 3])
```

### Example 3: CDC Pipeline
```python
# Simulate CDC events
cdc_events = [
{'user_id': 1, 'name': 'Alice', 'kind': 0}, # Insert
{'user_id': 2, 'name': 'Bob', 'kind': 0}, # Insert
{'user_id': 2, 'name': 'Bob_Updated', 'kind': 2},# Update
{'user_id': 3, 'name': 'Charlie', 'kind': 3}, # Delete
]

# Process CDC events
for event in cdc_events:
data = pa.Table.from_pydict({
'user_id': [event['user_id']],
'name': [event['name']],
'__row_kind__': pa.array([event['kind']], type=pa.int32())
})
write.write_arrow(data)
```

### Solution

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with paimon-python/pypaimon/write/writer/key_value_data_writer.py, especially _add_system_fields() and the proposed _extract_row_kind_column() and _deduplicate_by_primary_key() paths. Then inspect paimon-python/pypaimon/write/batch_table_write.py and its write_arrow and write_arrow_batch APIs. Done means RowKind values 0-3 are validated and preserved, the temporary column is removed, defaults remain INSERT, and the described unit coverage passes.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.