Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

When soft deleting, preserve previous values if we have them in memory #762

Merged
merged 7 commits into from
Jun 25, 2024
Merged

When soft deleting, preserve previous values if we have them in memory #762

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f544aa6
Preserve previous values from memory when soft-deleting a row that wa…
danafallon Jun 25, 2024
70648ec
Fix tests
danafallon Jun 25, 2024
37bfb8d
Protect against prevRow being a nil map
danafallon Jun 25, 2024
4caf734
Add tests, fix the nil map check
danafallon Jun 25, 2024
ab8aabe
Update rowData arg in tests; remove empty check
danafallon Jun 25, 2024
a88793b
Merge branch 'master' into dana/soft-delete-behavior
danafallon Jun 25, 2024
bb1825b
Merge branch 'master' into dana/soft-delete-behavior
danafallon Jun 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 16 additions & 9 deletions lib/optimization/table_data.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,17 +131,24 @@ func (t *TableData) InsertRow(pk string, rowData map[string]any, delete bool) {
prevRow, isOk := t.rowsData[pk]
if isOk {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(hide whitespace to make the diff simpler)

prevRowSize = size.GetApproxSize(prevRow)
for key, val := range rowData {
if val == constants.ToastUnavailableValuePlaceholder {
// Copy it from prevRow.
prevVal, isOk := prevRow[key]
if !isOk {
continue
if delete {
// If the row was deleted, preserve the previous values that we have in memory
rowData = prevRow
rowData[constants.DeleteColumnMarker] = true
} else {
for key, val := range rowData {
if val == constants.ToastUnavailableValuePlaceholder {
// Copy it from prevRow.
prevVal, isOk := prevRow[key]
if !isOk {
continue
}

// If we got back a TOASTED value, we need to use the previous row.
rowData[key] = prevVal
}

// If we got back a TOASTED value, we need to use the previous row.
rowData[key] = prevVal
}

}
}

Expand Down
28 changes: 24 additions & 4 deletions lib/optimization/table_data_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"testing"
"time"

"github.com/artie-labs/transfer/lib/config/constants"
"github.com/artie-labs/transfer/lib/typing/columns"

"github.com/artie-labs/transfer/lib/config"
Expand Down Expand Up @@ -208,7 +209,7 @@ func TestTableData_ContainsHardDeletes(t *testing.T) {
td := NewTableData(nil, config.Replication, nil, kafkalib.TopicConfig{}, "foo")
assert.Equal(t, 0, int(td.NumberOfRows()))

td.InsertRow("123", nil, true)
td.InsertRow("123", map[string]any{"id": "123"}, true)
assert.Equal(t, 1, int(td.NumberOfRows()))

assert.True(t, td.ContainsHardDeletes())
Expand All @@ -218,7 +219,7 @@ func TestTableData_ContainsHardDeletes(t *testing.T) {
td := NewTableData(nil, config.Replication, nil, kafkalib.TopicConfig{SoftDelete: true}, "foo")
assert.Equal(t, 0, int(td.NumberOfRows()))

td.InsertRow("123", nil, true)
td.InsertRow("123", map[string]any{"id": "123"}, true)
assert.Equal(t, 1, int(td.NumberOfRows()))
assert.False(t, td.ContainsHardDeletes())
}
Expand Down Expand Up @@ -270,12 +271,31 @@ func TestTableData_InsertRowIntegrity(t *testing.T) {
assert.False(t, td.ContainOtherOperations())

for i := 0; i < 100; i++ {
td.InsertRow("123", nil, true)
td.InsertRow("123", map[string]any{"id": "123"}, true)
assert.False(t, td.ContainOtherOperations())
}

for i := 0; i < 100; i++ {
td.InsertRow("123", nil, false)
td.InsertRow("123", map[string]any{"id": "123"}, false)
assert.True(t, td.ContainOtherOperations())
}
}

func TestTableData_InsertRowSoftDelete(t *testing.T) {
td := NewTableData(nil, config.Replication, nil, kafkalib.TopicConfig{SoftDelete: true}, "foo")
assert.Equal(t, 0, int(td.NumberOfRows()))

td.InsertRow("123", map[string]any{"id": "123", "name": "dana"}, false)
assert.Equal(t, 1, int(td.NumberOfRows()))
assert.Equal(t, "dana", td.Rows()[0]["name"])

td.InsertRow("123", map[string]any{"id": "123", "name": "dana2"}, false)
assert.Equal(t, 1, int(td.NumberOfRows()))
assert.Equal(t, "dana2", td.Rows()[0]["name"])

td.InsertRow("123", map[string]any{"id": "123", constants.DeleteColumnMarker: true}, true)
assert.Equal(t, 1, int(td.NumberOfRows()))
// The previous value should be preserved, along with the delete marker
assert.Equal(t, "dana2", td.Rows()[0]["name"])
assert.Equal(t, true, td.Rows()[0][constants.DeleteColumnMarker])
}