Hastyshell commented on code in PR #62256:
URL: https://github.com/apache/doris/pull/62256#discussion_r3061892990
##########
be/src/cloud/cloud_tablet.cpp:
##########
@@ -491,6 +491,34 @@ void CloudTablet::delete_rowsets(const
std::vector<RowsetSharedPtr>& to_delete,
_tablet_meta->modify_rs_metas({}, rs_metas, false);
}
+void CloudTablet::delete_rowsets_for_schema_change(const
std::vector<RowsetSharedPtr>& to_delete,
Review Comment:
Could we reuse the same abstraction as the local schema-change path here?
Local schema change already calls delete_rowsets(..., false) to directly remove
versions, so I wonder if cloud could extend delete_rowsets with the same
semantics rather than adding a dedicated delete_rowsets_for_schema_change
method.
##########
be/test/cloud/cloud_tablet_test.cpp:
##########
@@ -1361,4 +1361,252 @@ TEST_F(CloudTabletWarmUpStateTest,
TestWarmedUpOverridesNotWarmedUp) {
EXPECT_TRUE(_tablet->is_rowset_warmed_up(rowset->rowset_id()));
}
+class CloudTabletDeleteRowsetsForSchemaChangeTest : public testing::Test {
+public:
+ CloudTabletDeleteRowsetsForSchemaChangeTest() :
_engine(CloudStorageEngine(EngineOptions {})) {}
+
+ void SetUp() override {
+ _tablet_meta.reset(new TabletMeta(1, 2, 15673, 15674, 4, 5,
TTabletSchema(), 6, {{7, 8}},
+ UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F));
+ _tablet =
+ std::make_shared<CloudTablet>(_engine,
std::make_shared<TabletMeta>(*_tablet_meta));
+ }
+ void TearDown() override {}
+
+ RowsetSharedPtr create_rowset(Version version) {
+ auto rs_meta = std::make_shared<RowsetMeta>();
+ rs_meta->set_rowset_type(BETA_ROWSET);
+ rs_meta->set_version(version);
+ rs_meta->set_rowset_id(_engine.next_rowset_id());
+ RowsetSharedPtr rowset;
+ Status st = RowsetFactory::create_rowset(nullptr, "", rs_meta,
&rowset);
+ if (!st.ok()) {
+ return nullptr;
+ }
+ return rowset;
+ }
+
+protected:
+ TabletMetaSharedPtr _tablet_meta;
+ std::shared_ptr<CloudTablet> _tablet;
+ CloudStorageEngine _engine;
+};
+
+// Simulate the DORIS-25014 scenario:
+// - New tablet has compacted rowset [2-6] from compaction during SC
+// - SC produces individual output rowsets [2],[3],[4],[5],[6]
+// - Without the fix, add_rowsets fails to remove [2-6] because
+// [2].contains([2-6]) = false
+// - With delete_rowsets_for_schema_change, the stale compaction rowset is
+// removed from both _rs_version_map and version graph before add_rowsets
+TEST_F(CloudTabletDeleteRowsetsForSchemaChangeTest,
TestSchemaChangeDeletesCompactionRowset) {
+ // Setup: add placeholder [0-1] and compacted rowset [2-6]
+ auto rs_placeholder = create_rowset(Version(0, 1));
+ auto rs_compacted = create_rowset(Version(2, 6));
+ ASSERT_NE(rs_placeholder, nullptr);
+ ASSERT_NE(rs_compacted, nullptr);
+
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->add_rowsets({rs_placeholder, rs_compacted}, false, wlock,
false);
+ }
+ // Verify initial state
+ ASSERT_EQ(_tablet->rowset_map().size(), 2);
+ ASSERT_TRUE(_tablet->rowset_map().count(Version(2, 6)));
+
+ // SC produces individual rowsets
+ std::vector<RowsetSharedPtr> sc_output;
+ for (int v = 2; v <= 6; v++) {
+ auto rs = create_rowset(Version(v, v));
+ ASSERT_NE(rs, nullptr);
+ sc_output.push_back(rs);
+ }
+
+ // Simulate delete_rowsets_for_schema_change + add_rowsets
+ int64_t alter_version = 6;
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ // Collect rowsets in [2, alter_version]
+ std::vector<RowsetSharedPtr> to_delete;
+ for (auto& [v, rs] : _tablet->rowset_map()) {
+ if (v.first >= 2 && v.second <= alter_version) {
+ to_delete.push_back(rs);
+ }
+ }
+ ASSERT_EQ(to_delete.size(), 1); // only [2-6]
+ ASSERT_EQ(to_delete[0]->version(), Version(2, 6));
+
+ _tablet->delete_rowsets_for_schema_change(to_delete, wlock);
+
+ // [2-6] should be removed from rs_version_map
+ ASSERT_FALSE(_tablet->rowset_map().count(Version(2, 6)));
+ // Should NOT go to stale (to avoid stale path conflicts), but to
unused
+ ASSERT_FALSE(_tablet->has_stale_rowsets());
+ ASSERT_TRUE(_tablet->need_remove_unused_rowsets());
+
+ _tablet->add_rowsets(std::move(sc_output), false, wlock, false);
+ }
+
+ // Verify: individual SC rowsets are now in rs_version_map
+ ASSERT_EQ(_tablet->rowset_map().size(), 6); // [0-1] + 5 individual
+ for (int v = 2; v <= 6; v++) {
+ ASSERT_TRUE(_tablet->rowset_map().count(Version(v, v)))
+ << "Missing version " << v << "-" << v;
+ }
+ ASSERT_FALSE(_tablet->rowset_map().count(Version(2, 6)));
+
+ // Verify: capture_consistent_versions works correctly (no stale edges)
+ auto versions_result =
_tablet->capture_consistent_versions_unlocked(Version(0, 6), {});
+ ASSERT_TRUE(versions_result.has_value()) << versions_result.error();
+ auto& versions = versions_result.value();
+ ASSERT_EQ(versions.size(), 6); // [0-1] + [2],[3],[4],[5],[6]
+ ASSERT_EQ(versions[0], Version(0, 1));
+ for (int i = 0; i < 5; i++) {
+ ASSERT_EQ(versions[i + 1], Version(2 + i, 2 + i));
+ }
+}
+
+// Test that delete_rowsets_for_schema_change with empty input is a no-op
+TEST_F(CloudTabletDeleteRowsetsForSchemaChangeTest, TestEmptyDeleteIsNoop) {
+ auto rs = create_rowset(Version(0, 1));
+ ASSERT_NE(rs, nullptr);
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->add_rowsets({rs}, false, wlock, false);
+ }
+ ASSERT_EQ(_tablet->rowset_map().size(), 1);
+
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->delete_rowsets_for_schema_change({}, wlock);
+ }
+ ASSERT_EQ(_tablet->rowset_map().size(), 1);
+ ASSERT_FALSE(_tablet->has_stale_rowsets());
+}
+
+// Test with multiple compaction rowsets spanning different version ranges
+TEST_F(CloudTabletDeleteRowsetsForSchemaChangeTest,
TestMultipleCompactionRowsets) {
+ auto rs_placeholder = create_rowset(Version(0, 1));
+ auto rs_comp1 = create_rowset(Version(2, 5));
+ auto rs_comp2 = create_rowset(Version(6, 10));
+ auto rs_post = create_rowset(Version(11, 11)); // after alter_version,
should NOT be deleted
+ ASSERT_NE(rs_placeholder, nullptr);
+ ASSERT_NE(rs_comp1, nullptr);
+ ASSERT_NE(rs_comp2, nullptr);
+ ASSERT_NE(rs_post, nullptr);
+
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->add_rowsets({rs_placeholder, rs_comp1, rs_comp2, rs_post},
false, wlock, false);
+ }
+ ASSERT_EQ(_tablet->rowset_map().size(), 4);
+
+ // SC output: individual rowsets for versions 2-10
+ std::vector<RowsetSharedPtr> sc_output;
+ for (int v = 2; v <= 10; v++) {
+ auto rs = create_rowset(Version(v, v));
+ ASSERT_NE(rs, nullptr);
+ sc_output.push_back(rs);
+ }
+
+ int64_t alter_version = 10;
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ std::vector<RowsetSharedPtr> to_delete;
+ for (auto& [v, rs] : _tablet->rowset_map()) {
+ if (v.first >= 2 && v.second <= alter_version) {
+ to_delete.push_back(rs);
+ }
+ }
+ ASSERT_EQ(to_delete.size(), 2); // [2-5] and [6-10]
+
+ _tablet->delete_rowsets_for_schema_change(to_delete, wlock);
+
+ // Post-alter rowset should survive
+ ASSERT_TRUE(_tablet->rowset_map().count(Version(11, 11)));
+ ASSERT_FALSE(_tablet->rowset_map().count(Version(2, 5)));
+ ASSERT_FALSE(_tablet->rowset_map().count(Version(6, 10)));
+
+ _tablet->add_rowsets(std::move(sc_output), false, wlock, false);
+ }
+
+ // Verify: [0-1], [2],[3],...,[10], [11-11]
+ ASSERT_EQ(_tablet->rowset_map().size(), 11);
+
+ // Verify capture
+ auto versions_result =
_tablet->capture_consistent_versions_unlocked(Version(0, 11), {});
+ ASSERT_TRUE(versions_result.has_value()) << versions_result.error();
+ auto& versions = versions_result.value();
+ ASSERT_EQ(versions.size(), 11);
+ ASSERT_EQ(versions[0], Version(0, 1));
+ for (int i = 0; i < 9; i++) {
+ ASSERT_EQ(versions[i + 1], Version(2 + i, 2 + i));
+ }
+ ASSERT_EQ(versions[10], Version(11, 11));
+}
+
+// Reproduce the CI crash scenario: SC delete puts rowsets to stale, then
+// compaction creates a new stale path with overlapping version keys. When
+// one stale path is cleaned, the other hits DCHECK(false) because the
+// version is already removed from _stale_rs_version_map.
+// With the fix (bypassing stale tracking), this should not happen.
+TEST_F(CloudTabletDeleteRowsetsForSchemaChangeTest,
TestNoStalePathConflictWithCompaction) {
+ // Setup: [0-1] placeholder, [2-6] compaction product during SC
+ auto rs_placeholder = create_rowset(Version(0, 1));
+ auto rs_compacted = create_rowset(Version(2, 6));
+ ASSERT_NE(rs_placeholder, nullptr);
+ ASSERT_NE(rs_compacted, nullptr);
+
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->add_rowsets({rs_placeholder, rs_compacted}, false, wlock,
false);
+ }
+
+ // SC output: individual rowsets [2],[3],[4],[5],[6]
+ std::vector<RowsetSharedPtr> sc_output;
+ for (int v = 2; v <= 6; v++) {
+ sc_output.push_back(create_rowset(Version(v, v)));
+ }
+
+ // Step 1: delete_rowsets_for_schema_change + add SC output
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ _tablet->delete_rowsets_for_schema_change({rs_compacted}, wlock);
+ _tablet->add_rowsets(std::move(sc_output), false, wlock, false);
+ }
+ // Stale should be empty — SC delete bypasses stale tracking
+ ASSERT_FALSE(_tablet->has_stale_rowsets());
+
+ // Step 2: compaction merges SC output [2],[3],[4],[5],[6] -> [2-6]
+ auto rs_new_compacted = create_rowset(Version(2, 6));
+ std::vector<RowsetSharedPtr> compaction_input;
+ {
+ std::unique_lock wlock(_tablet->get_header_lock());
+ for (auto& [v, rs] : _tablet->rowset_map()) {
+ if (v.first >= 2 && v.second <= 6) {
+ compaction_input.push_back(rs);
+ }
+ }
+ ASSERT_EQ(compaction_input.size(), 5);
+ // Normal compaction delete_rowsets — this WILL use stale tracking
+ _tablet->delete_rowsets(compaction_input, wlock);
+ _tablet->add_rowsets({rs_new_compacted}, false, wlock, false);
+ }
+ // Now stale has the compaction inputs
+ ASSERT_TRUE(_tablet->has_stale_rowsets());
+
+ // Step 3: delete_expired_stale_rowsets — this is where CI crashed
+ // With old code: stale path from SC and compaction both reference [2-6]
key,
+ // causing DCHECK(false). With fix: only compaction stale path exists, no
conflict.
+ config::tablet_rowset_stale_sweep_time_sec = 0; // expire immediately
Review Comment:
shall we add a guard to restore this config
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
