[BugFix] fix bad sst when using cloud native pk index (backport #53334)…

… (#53339)

Co-authored-by: Yixin Luo <[email protected]>

be/src/storage/lake/lake_persistent_index.cpp

@@ -503,7 +503,7 @@ Status LakePersistentIndex::major_compact(TabletManager* tablet_mgr, const Table
 }
 
 Status LakePersistentIndex::apply_opcompaction(const TxnLogPB_OpCompaction& op_compaction) {
-    if (op_compaction.input_sstables().empty()) {
+    if (op_compaction.input_sstables().empty() || !op_compaction.has_output_sstable()) {
         return Status::OK();
     }
 

be/src/storage/sstable/table_builder.cpp

@@ -15,6 +15,7 @@
 #include "storage/sstable/filter_block.h"
 #include "storage/sstable/filter_policy.h"
 #include "storage/sstable/format.h"
+#include "testutil/sync_point.h"
 #include "util/crc32c.h"
 #include "util/slice.h"
 
@@ -241,6 +242,8 @@ Status TableBuilder::Finish() {
         WriteBlock(&r->index_block, &index_block_handle);
     }
 
+    TEST_SYNC_POINT_CALLBACK("table_builder_footer_error", &r->status);
+
     // Write footer
     if (ok()) {
         Footer footer;
@@ -254,7 +257,9 @@ Status TableBuilder::Finish() {
         }
     }
     // sync file at last
-    r->status = r->file->sync();
+    if (ok()) {
+        r->status = r->file->sync();
+    }
     return r->status;
 }
 

be/test/storage/lake/persistent_index_sstable_test.cpp

@@ -438,4 +438,37 @@ TEST_F(PersistentIndexSstableTest, test_index_value_protobuf) {
     }
 }
 
+TEST_F(PersistentIndexSstableTest, test_ioerror_inject) {
+    const int N = 10000;
+    sstable::Options options;
+    const std::string filename = "test_ioerror_inject.sst";
+    ASSIGN_OR_ABORT(auto file, fs::new_writable_file(lake::join_path(kTestDir, filename)));
+    sstable::TableBuilder builder(options, file.get());
+    for (int i = 0; i < N; i++) {
+        std::string str = fmt::format("test_key_{:016X}", i);
+        IndexValue val(i);
+        builder.Add(Slice(str), Slice(val.v, 8));
+    }
+    SyncPoint::GetInstance()->SetCallBack("table_builder_footer_error",
+                                          [&](void* arg) { *(Status*)arg = Status::IOError("ut_test"); });
+    SyncPoint::GetInstance()->EnableProcessing();
+    auto st = builder.Finish();
+    SyncPoint::GetInstance()->ClearCallBack("table_builder_footer_error");
+    SyncPoint::GetInstance()->DisableProcessing();
+    uint64_t filesz = builder.FileSize();
+    if (st.ok()) {
+        // scan & check
+        sstable::Table* sstable = nullptr;
+        ASSIGN_OR_ABORT(auto read_file, fs::new_random_access_file(lake::join_path(kTestDir, filename)));
+        CHECK_OK(sstable::Table::Open(options, read_file.get(), filesz, &sstable));
+        sstable::ReadOptions read_options;
+        sstable::Iterator* iter = sstable->NewIterator(read_options);
+        for (iter->SeekToFirst(); iter->Valid() && iter->status().ok(); iter->Next()) {
+        }
+        ASSERT_TRUE(iter->status().ok());
+        delete iter;
+        delete sstable;
+    }
+}
+
 } // namespace starrocks::lake