Skip to content

Commit 087b1a7

Browse files
author
luohaha
committed
[Enhancement] Make the unsort SST writer op-aware for separate-sort-key mixed upsert/delete loads
When lake_enable_pk_preserve_txn_delete_order is on (now the default), a load that carries a hidden __op column keeps DELETE and UPSERT rows distinct and relies on the spill merge REPLACE-aggregating each PK to its latest op before splitting the DELETEs into a del file. For a separate-sort-key table the merge is ordered by the sort key, so same-PK rows are not adjacent and aggregation is disabled -- a DELETE followed by a later UPSERT of the same key would then leak the DELETE to the del file and erase the (correct) UPSERT. Resolve op order in the unsort SST writer instead, using the per-row rssid_rowid that already flows through this path as the ordering key: - DELETE rows are fed via TabletWriter::append_pk_index_deletes -> PkTabletUnsortSSTWriter::append_delete_records; they carry order but no segment rowid (kDeleteRowid) and do not advance the rowid counter. - reconcile_entry keeps the largest-order row per PK across upserts and deletes; a losing UPSERT's rowid goes to the delete vector, a losing/superseded DELETE is dropped. - At flush, a PK whose latest op is a DELETE is written as a persistent-index tombstone (rssid == rowid == UINT32_MAX), which ingest preserves through the shared_rssid projection (is_index_tombstone) and publish applies as an erase -- no del file and no del-file key-format mismatch. Both the in-memory and the intermediate-spill k-way-merge paths emit tombstones. - The parallel-merge task routes deletes to the writer only for separate-sort-key (sort-key == PK keeps the existing del-file split) and still writes a 0-row anchor segment for a delete-only batch so the SST stays 1:1 with segments. append_records now returns early on an empty chunk, so the 0-row anchor write no longer trips the "requires per-row order keys" check. Add a BE UT (test_unsort_sst_writer_op_aware_reconcile) covering delete-wins, upsert-wins (superseded delete), delete-only and plain-upsert keys, asserting both the delete vector and that delete-winners are tombstones in the SST. Signed-off-by: luohaha <casey.luo@celerdata.com>
1 parent 6584e51 commit 087b1a7

8 files changed

Lines changed: 222 additions & 36 deletions

‎be/src/storage/lake/pk_tablet_sst_writer.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,12 @@ class DefaultSSTWriter {
4545
const std::vector<uint32_t>* column_indexes = nullptr) {
4646
return Status::OK();
4747
}
48+
// Feed DELETE rows to writers that resolve op order themselves (the separate-sort-key unsort
49+
// writer). Default is a no-op: other writers receive deletes via the caller's del-file path.
50+
virtual Status append_delete_records(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
51+
const std::vector<uint32_t>* column_indexes = nullptr) {
52+
return Status::OK();
53+
}
4854
virtual Status reset_sst_writer(const std::shared_ptr<LocationProvider>& location_provider,
4955
const std::shared_ptr<FileSystem>& fs) {
5056
return Status::OK();

‎be/src/storage/lake/pk_tablet_unsort_sst_writer.cpp‎

Lines changed: 62 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -66,10 +66,43 @@ Status PkTabletUnsortSSTWriter::reset_sst_writer(const std::shared_ptr<LocationP
6666

6767
Status PkTabletUnsortSSTWriter::append_sst_record(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
6868
const std::vector<uint32_t>* column_indexes) {
69+
return append_records(data, rssid_rowids, column_indexes, /*is_delete=*/false);
70+
}
71+
72+
Status PkTabletUnsortSSTWriter::append_delete_records(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
73+
const std::vector<uint32_t>* column_indexes) {
74+
return append_records(data, rssid_rowids, column_indexes, /*is_delete=*/true);
75+
}
76+
77+
void PkTabletUnsortSSTWriter::reconcile_entry(std::string key, uint64_t order, uint32_t rowid) {
78+
// Keep the row with the largest order (last op wins). When an UPSERT loses (its rowid is a real
79+
// segment position, not kDeleteRowid), record it so publish masks it with a delete vector. A
80+
// DELETE has no segment row (kDeleteRowid), so a losing/superseded DELETE contributes nothing.
81+
auto it = _map.find(key);
82+
if (it == _map.end()) {
83+
// Rough per-entry footprint: encoded key bytes + value + btree node overhead.
84+
static constexpr size_t kBtreeEntryOverhead = 24;
85+
_map_mem_usage += key.size() + sizeof(Entry) + kBtreeEntryOverhead;
86+
_map.emplace(std::move(key), Entry{order, rowid});
87+
} else if (order > it->second.order) {
88+
if (it->second.rowid != kDeleteRowid) {
89+
_deleted_rowids.push_back(it->second.rowid);
90+
}
91+
it->second = Entry{order, rowid};
92+
} else if (rowid != kDeleteRowid) {
93+
_deleted_rowids.push_back(rowid);
94+
}
95+
}
96+
97+
Status PkTabletUnsortSSTWriter::append_records(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
98+
const std::vector<uint32_t>* column_indexes, bool is_delete) {
6999
if (_wf == nullptr) {
70100
return Status::InternalError("unsort pk sst writer not initialized");
71101
}
72102
const size_t num_rows = data.num_rows();
103+
if (num_rows == 0) {
104+
return Status::OK();
105+
}
73106
if (rssid_rowids == nullptr || rssid_rowids->size() != num_rows) {
74107
return Status::InternalError(fmt::format("unsort pk sst writer requires per-row order keys, expected {} got {}",
75108
num_rows, rssid_rowids == nullptr ? 0 : rssid_rowids->size()));
@@ -90,24 +123,13 @@ Status PkTabletUnsortSSTWriter::append_sst_record(const Chunk& data, const std::
90123
ASSIGN_OR_RETURN(const Slice* vkeys, encode_pk_keys(*pk_data, &keys, &owned_column));
91124
for (size_t i = 0; i < num_rows; i++) {
92125
const uint64_t order = (*rssid_rowids)[i];
93-
const uint32_t rowid = _next_rowid + static_cast<uint32_t>(i);
94-
// Insert; on duplicate primary key keep the row with the larger order (last-flushed wins),
95-
// and record the loser's segment rowid so publish masks it with a delete vector.
96-
std::string key = vkeys[i].to_string();
97-
auto it = _map.find(key);
98-
if (it == _map.end()) {
99-
// Rough per-entry footprint: encoded key bytes + value + btree node overhead.
100-
static constexpr size_t kBtreeEntryOverhead = 24;
101-
_map_mem_usage += key.size() + sizeof(Entry) + kBtreeEntryOverhead;
102-
_map.emplace(std::move(key), Entry{order, rowid});
103-
} else if (order > it->second.order) {
104-
_deleted_rowids.push_back(it->second.rowid);
105-
it->second = Entry{order, rowid};
106-
} else {
107-
_deleted_rowids.push_back(rowid);
108-
}
126+
// DELETE rows occupy no segment position; they only carry order for the per-PK reconciliation.
127+
const uint32_t rowid = is_delete ? kDeleteRowid : (_next_rowid + static_cast<uint32_t>(i));
128+
reconcile_entry(vkeys[i].to_string(), order, rowid);
129+
}
130+
if (!is_delete) {
131+
_next_rowid += static_cast<uint32_t>(num_rows);
109132
}
110-
_next_rowid += static_cast<uint32_t>(num_rows);
111133
// Bound memory: once the map reaches the memtable threshold, spill it to an intermediate SST.
112134
// The residual map plus all intermediates are k-way merged into the final SST at flush time.
113135
if (is_map_full()) {
@@ -249,19 +271,29 @@ Status PkTabletUnsortSSTWriter::merge_intermediates_into(sstable::TableBuilder*
249271
const uint64_t order = static_cast<uint64_t>(index_value_pb.values(0).version());
250272
const uint32_t rowid = index_value_pb.values(0).rowid();
251273
if (!has_best || order > best_order) {
252-
if (has_best) {
274+
// A losing UPSERT (real rowid) is masked by the delete vector; a losing DELETE
275+
// (kDeleteRowid) has no segment row and is simply dropped.
276+
if (has_best && best_rowid != kDeleteRowid) {
253277
_deleted_rowids.push_back(best_rowid);
254278
}
255279
has_best = true;
256280
best_order = order;
257281
best_rowid = rowid;
258-
} else {
282+
} else if (rowid != kDeleteRowid) {
259283
_deleted_rowids.push_back(rowid);
260284
}
261285
merged->Next();
262286
}
263287
IndexValuesWithVerPB out;
264-
out.add_values()->set_rowid(best_rowid);
288+
auto* value = out.add_values();
289+
if (best_rowid == kDeleteRowid) {
290+
// Latest op for this key is a DELETE: write a persistent-index tombstone (see the
291+
// in-memory path in flush_sst_writer) so publish erases the key.
292+
value->set_rssid(kDeleteRowid);
293+
value->set_rowid(kDeleteRowid);
294+
} else {
295+
value->set_rowid(best_rowid);
296+
}
265297
RETURN_IF_ERROR(builder->Add(Slice(key), Slice(out.SerializeAsString())));
266298
}
267299
return merged->status();
@@ -282,7 +314,16 @@ StatusOr<std::pair<FileInfo, PersistentIndexSstableRangePB>> PkTabletUnsortSSTWr
282314
// from shared_rssid/shared_version at ingest time (single-segment SST).
283315
for (const auto& [k, v] : _map) {
284316
IndexValuesWithVerPB index_value_pb;
285-
index_value_pb.add_values()->set_rowid(v.rowid);
317+
auto* value = index_value_pb.add_values();
318+
if (v.rowid == kDeleteRowid) {
319+
// Latest op for this key is a DELETE: write a persistent-index tombstone
320+
// (rssid == rowid == UINT32_MAX). Ingest preserves the sentinel through shared_rssid
321+
// projection (is_index_tombstone) and publish applies it as an erase of the key.
322+
value->set_rssid(kDeleteRowid);
323+
value->set_rowid(kDeleteRowid);
324+
} else {
325+
value->set_rowid(v.rowid);
326+
}
286327
RETURN_IF_ERROR(builder.Add(Slice(k), Slice(index_value_pb.SerializeAsString())));
287328
}
288329
} else {

‎be/src/storage/lake/pk_tablet_unsort_sst_writer.h‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
#pragma once
1616

17+
#include <limits>
1718
#include <memory>
1819
#include <string>
1920
#include <utility>
@@ -46,15 +47,28 @@ class PkTabletUnsortSSTWriter : public PkTabletSSTWriter {
4647
~PkTabletUnsortSSTWriter() override = default;
4748
Status append_sst_record(const Chunk& data, const std::vector<uint64_t>* rssid_rowids = nullptr,
4849
const std::vector<uint32_t>* column_indexes = nullptr) override;
50+
// Feed the batch's DELETE rows (op-aware separate-sort-key path). They carry a per-row ordering key
51+
// (rssid_rowids) but no segment row, so they participate in the per-PK last-op reconciliation
52+
// without occupying a rowid. A DELETE that is the latest op for its PK is written into the SST as a
53+
// persistent-index tombstone; a DELETE superseded by a later UPSERT is dropped.
54+
Status append_delete_records(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
55+
const std::vector<uint32_t>* column_indexes = nullptr) override;
4956
Status reset_sst_writer(const std::shared_ptr<LocationProvider>& location_provider,
5057
const std::shared_ptr<FileSystem>& fs) override;
5158
StatusOr<std::pair<FileInfo, PersistentIndexSstableRangePB>> flush_sst_writer() override;
5259
bool has_file_info() const override { return _wf != nullptr; }
5360
std::vector<uint32_t> take_deleted_rowids() override { return std::move(_deleted_rowids); }
5461

5562
private:
56-
// encoded primary key -> {order, rowid}. `order` is the per-row rssid_rowid used only to pick the
57-
// dedup winner (kept the largest); `rowid` is the position within the current segment.
63+
// A row that has no segment position (a DELETE) stores this reserved rowid, matching the
64+
// UINT32_MAX delete-rowid convention used elsewhere in the PK write path. At flush a winning
65+
// entry with this rowid is emitted into the SST as a persistent-index tombstone
66+
// (rssid == rowid == UINT32_MAX) instead of a PK -> rowid mapping, so publish erases the key.
67+
static constexpr uint32_t kDeleteRowid = std::numeric_limits<uint32_t>::max();
68+
69+
// encoded primary key -> {order, rowid}. `order` is the per-row rssid_rowid used to pick the
70+
// last-op winner (kept the largest); `rowid` is the position within the current segment, or
71+
// kDeleteRowid when the winning op is a DELETE.
5872
struct Entry {
5973
uint64_t order;
6074
uint32_t rowid;
@@ -72,6 +86,13 @@ class PkTabletUnsortSSTWriter : public PkTabletSSTWriter {
7286
// columns are those with a tablet column id < num_key_columns. `out` then has the PK columns at
7387
// positions [0, num_key), which is what encode_pk_keys expects.
7488
Status project_pk_columns(const Chunk& data, const std::vector<uint32_t>& column_indexes, Chunk* out) const;
89+
// Shared body of append_sst_record (is_delete=false) and append_delete_records (is_delete=true):
90+
// encode each row's PK and reconcile it into `_map` by rssid_rowid order (last op wins). UPSERTs
91+
// take a running segment rowid; DELETEs use kDeleteRowid and do not advance the rowid counter.
92+
Status append_records(const Chunk& data, const std::vector<uint64_t>* rssid_rowids,
93+
const std::vector<uint32_t>* column_indexes, bool is_delete);
94+
// Reconcile one row (already encoded) into `_map`, masking the loser's segment rowid (if any).
95+
void reconcile_entry(std::string key, uint64_t order, uint32_t rowid);
7596
// Whether `_map` has reached the memtable memory threshold and should be spilled to an SST.
7697
bool is_map_full() const;
7798
// Spill the current `_map` to a new intermediate SST (storing order+rowid), then clear the map.

‎be/src/storage/lake/pk_tablet_writer.cpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,13 @@ Status HorizontalPkTabletWriter::write(const Chunk& data, SegmentPB* segment, bo
6666
return Status::OK();
6767
}
6868

69+
Status HorizontalPkTabletWriter::append_pk_index_deletes(const Chunk& data, const std::vector<uint64_t>& rssid_rowids) {
70+
// DELETE rows are only fed to the eager PK-index SST writer (to resolve the latest op per key);
71+
// they are NOT written to a segment. Only the unsort SST writer acts on them; other SST writers
72+
// no-op (deletes reach them via the caller's del file instead).
73+
return _pk_sst_writer->append_delete_records(data, &rssid_rowids);
74+
}
75+
6976
Status HorizontalPkTabletWriter::flush_del_file(const Column& deletes, uint32_t op_offset) {
7077
auto name = gen_del_filename(_txn_id);
7178
WritableFileOptions wopts;

‎be/src/storage/lake/pk_tablet_writer.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ class HorizontalPkTabletWriter : public HorizontalGeneralTabletWriter {
4444
DISALLOW_COPY(HorizontalPkTabletWriter);
4545

4646
Status write(const Chunk& data, const std::vector<uint64_t>& rssid_rowids, SegmentPB* segment = nullptr) override;
47+
Status append_pk_index_deletes(const Chunk& data, const std::vector<uint64_t>& rssid_rowids) override;
4748

4849
Status write(const Chunk& data, SegmentPB* segment = nullptr, bool eos = false) override;
4950

‎be/src/storage/lake/tablet_internal_parallel_merge_task.cpp‎

Lines changed: 42 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ namespace {
6666
Status write_one_merged_chunk(TabletWriter* writer, Chunk* chunk, bool has_op, const Schema& merge_schema,
6767
const Schema& write_schema, const std::vector<size_t>& char_field_indexes,
6868
PrimaryKeyEncodingType pk_enc, MutableColumnPtr* deletes, bool* wrote_upsert,
69-
const std::vector<uint64_t>* rssid_rowids) {
69+
bool* had_deletes, const std::vector<uint64_t>* rssid_rowids) {
7070
if (!has_op) {
7171
ChunkHelper::padding_char_columns(char_field_indexes, write_schema, writer->tablet_schema(), chunk);
7272
// Separate-sort-key unsort path forwards per-row ordering keys so the writer can resolve
@@ -88,13 +88,34 @@ Status write_one_merged_chunk(TabletWriter* writer, Chunk* chunk, bool has_op, c
8888
for (uint32_t i = 0; i < nrows; i++) {
8989
(ops[i] == TOpType::UPSERT ? up_idx : del_idx).push_back(i);
9090
}
91-
// Encode net-deleted keys.
91+
// Net-deleted keys.
9292
if (!del_idx.empty()) {
93-
if (*deletes == nullptr) {
94-
RETURN_IF_ERROR(PrimaryKeyEncoder::create_column(merge_schema, deletes, pk_enc));
93+
*had_deletes = true;
94+
if (writer->tablet_schema()->has_separate_sort_key()) {
95+
// Separate-sort-key unsort path: the merge is ordered by sort key, so a DELETE and a later
96+
// UPSERT of the same PK are not adjacent and cannot be REPLACE-aggregated to the latest op
97+
// before this split. Instead feed the DELETE rows (with their per-row ordering keys) to the
98+
// eager SST writer, which reconciles them against the upserts by order (last op wins) and
99+
// emits a persistent-index tombstone for a PK whose latest op is a DELETE. rssid_rowids is
100+
// always present on this path.
101+
DCHECK(rssid_rowids != nullptr);
102+
auto del_chunk = chunk->clone_empty_with_schema(del_idx.size());
103+
del_chunk->append_selective(*chunk, del_idx.data(), 0, del_idx.size());
104+
std::vector<uint64_t> del_rssid_rowids;
105+
del_rssid_rowids.reserve(del_idx.size());
106+
for (uint32_t i : del_idx) {
107+
del_rssid_rowids.push_back((*rssid_rowids)[i]);
108+
}
109+
RETURN_IF_ERROR(writer->append_pk_index_deletes(*del_chunk, del_rssid_rowids));
110+
} else {
111+
// sort-key == PK: the merge already reduced each PK to its latest op, so all DELETEs here
112+
// are net deletes; encode them for this batch's del file.
113+
if (*deletes == nullptr) {
114+
RETURN_IF_ERROR(PrimaryKeyEncoder::create_column(merge_schema, deletes, pk_enc));
115+
}
116+
PrimaryKeyEncoder::encode_selective(merge_schema, *chunk, del_idx.data(), del_idx.size(), deletes->get(),
117+
pk_enc);
95118
}
96-
PrimaryKeyEncoder::encode_selective(merge_schema, *chunk, del_idx.data(), del_idx.size(), deletes->get(),
97-
pk_enc);
98119
}
99120
// Write upsert rows as a chunk without __op. Copy the upsert rows into a temp chunk (Chunk-level
100121
// append_selective handles the immutable columns), then repackage its leading data columns into a
@@ -131,10 +152,13 @@ Status write_one_merged_chunk(TabletWriter* writer, Chunk* chunk, bool has_op, c
131152
// sort after, and wrongly erase, a key re-upserted in that later segment. Mirrors the serial path, which
132153
// writes a 0-row segment for a delete-only flush.
133154
Status finalize_merged_batch(TabletWriter* writer, bool has_op, const Schema& write_schema,
134-
const MutableColumnPtr& deletes, bool wrote_upsert,
155+
const MutableColumnPtr& deletes, bool wrote_upsert, bool had_deletes,
135156
RuntimeProfile::Counter* write_io_timer) {
136-
const bool has_deletes = has_op && deletes != nullptr && deletes->size() > 0;
137-
if (has_deletes && !wrote_upsert) {
157+
// A batch that produced deletes but no upsert segment still needs a 0-row anchor segment so its
158+
// SST (and, for the sort-key==PK path, its del file) attaches to a real segment index rather than
159+
// colliding with a later batch's segment 0. `had_deletes` covers both delete sinks: the del-file
160+
// column (sort-key==PK) and the eager SST tombstones (separate-sort-key).
161+
if (has_op && had_deletes && !wrote_upsert) {
138162
SCOPED_TIMER(write_io_timer);
139163
auto empty_chunk = ChunkFactory::new_chunk(write_schema, 0);
140164
RETURN_IF_ERROR(writer->write(*empty_chunk, nullptr));
@@ -143,7 +167,9 @@ Status finalize_merged_batch(TabletWriter* writer, bool has_op, const Schema& wr
143167
SCOPED_TIMER(write_io_timer);
144168
RETURN_IF_ERROR(writer->flush());
145169
}
146-
if (has_deletes) {
170+
// Only the sort-key==PK path routes deletes to a del file; the separate-sort-key path records them
171+
// as tombstones inside the eager SST during flush() above, so there is no del file to write here.
172+
if (deletes != nullptr && deletes->size() > 0) {
147173
SCOPED_TIMER(write_io_timer);
148174
// op_offset is a placeholder here; the real (global) value is assigned in merge_other_writer when
149175
// this batch's writer is consolidated into the parent, based on the cumulative segment count.
@@ -215,8 +241,9 @@ void TabletInternalParallelMergeTask::run() {
215241
// Per-row ordering keys for the separate-sort-key unsort path (empty otherwise); pulled alongside
216242
// each chunk so the writer resolves duplicate primary keys by last-flushed-wins.
217243
std::vector<uint64_t> rssid_rowids;
218-
MutableColumnPtr deletes; // accumulates net-deleted keys for this batch's del file
244+
MutableColumnPtr deletes; // accumulates net-deleted keys for this batch's del file (sort-key==PK)
219245
bool wrote_upsert = false; // whether this batch produced any upsert segment
246+
bool had_deletes = false; // whether this batch saw any DELETE rows (either delete sink)
220247
// CANCELLATION CHECK: read and write each merged chunk. _quit_flag (when non-null) lets a sibling
221248
// task's error abort the loop early; when nullptr, cancellation is unsupported and we run to completion.
222249
while (_quit_flag == nullptr || !_quit_flag->load()) {
@@ -240,13 +267,15 @@ void TabletInternalParallelMergeTask::run() {
240267
}
241268
SCOPED_TIMER(_write_io_timer);
242269
st = write_one_merged_chunk(_writer.get(), chunk, has_op, *_schema, write_schema, char_field_indexes, pk_enc,
243-
&deletes, &wrote_upsert, _need_rssid_rowids ? &rssid_rowids : nullptr);
270+
&deletes, &wrote_upsert, &had_deletes,
271+
_need_rssid_rowids ? &rssid_rowids : nullptr);
244272
if (!st.ok()) {
245273
break;
246274
}
247275
}
248276
if (st.ok()) {
249-
st = finalize_merged_batch(_writer.get(), has_op, write_schema, deletes, wrote_upsert, _write_io_timer);
277+
st = finalize_merged_batch(_writer.get(), has_op, write_schema, deletes, wrote_upsert, had_deletes,
278+
_write_io_timer);
250279
}
251280
if (st.ok()) {
252281
hot_delete_merged_spill_files(_writer.get(), _task.get());

0 commit comments

Comments
 (0)