[fix](orc) handle cancellation during condition cache seek (#67141)

### What problem does this PR solve?

Issue Number: DORIS-28199

Problem Summary:

Condition-cache granule skipping can call ORC seekToRow before the
existing nextBatch exception boundary. When a query is cancelled during
that I/O, the ORC input stream throws a stop exception that escapes
get_block and may terminate the BE process.

### Release note

Fix BE process termination when an ORC scan is cancelled during
condition-cache seeking.

### Check List (For Author)

- Test
    - [x] Regression test
    - [x] Unit Test
    - [ ] Manual test
    - [ ] No need to test or manual test.

Verification: NewOrcReaderTest.* passed 163/163 under ASAN_UT.
Clang-format 16 passed for both affected C++ files.

- Behavior changed:
    - [ ] No.
- [x] Yes. Cancellation during a condition-cache seek now returns clean
EOF instead of allowing the ORC stop exception to escape.

- Does this need documentation?
    - [x] No.
    - [ ] Yes.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

Author: Gabriel39

be/src/format_v2/orc/orc_reader.cpp

@@ -1751,6 +1751,8 @@ void OrcReader::_skip_condition_cache_false_granules(size_t* rows, bool* eof) {
     }
     if (target_row > _state->condition_cache_next_row) {
         DORIS_CHECK(target_row <= file_total_rows);
+        DBUG_EXECUTE_IF("OrcReader._skip_condition_cache_false_granules.before_seek_to_row",
+                        DBUG_RUN_CALLBACK());
         _state->row_reader->seekToRow(target_row);
         if (_io_ctx != nullptr) {
             _io_ctx->condition_cache_filtered_rows += target_row - _state->condition_cache_next_row;
@@ -1939,11 +1941,13 @@ Status OrcReader::get_block(Block* file_block, size_t* rows, bool* eof) {
 
     bool has_next = false;
     while (true) {
-        _skip_condition_cache_false_granules(rows, eof);
-        if (*eof) {
-            return Status::OK();
-        }
         try {
+            // Condition-cache seeks can perform I/O, so keep them in the same cancellation
+            // boundary as next().
+            _skip_condition_cache_false_granules(rows, eof);
+            if (*eof) {
+                return Status::OK();
+            }
             _state->orc_lazy_selection_valid = false;
             _state->orc_lazy_selected_rows.clear();
             _state->orc_lazy_input_rows = 0;

be/test/format_v2/orc/orc_reader_test.cpp

@@ -6133,6 +6133,54 @@ TEST_F(NewOrcReaderTest, ConditionCacheHitSkipsFalseGranulesBeforeColumnRead) {
     EXPECT_EQ(rows, 0);
 }
 
+TEST_F(NewOrcReaderTest, ConditionCacheSeekReturnsCleanEofWhenCancelled) {
+    constexpr int64_t row_count = ConditionCacheContext::GRANULE_SIZE * 2;
+    const auto file_path = (_test_dir / "condition_cache_cancelled_seek.orc").string();
+    write_large_orc_int_file(file_path, row_count);
+
+    auto io_ctx = std::make_shared<io::IOContext>();
+    auto reader = create_reader_for_path(file_path, nullptr, io_ctx);
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    ASSERT_TRUE(reader->init(&state).ok());
+
+    std::vector<format::ColumnDefinition> schema;
+    ASSERT_TRUE(reader->get_schema(&schema).ok());
+    ASSERT_EQ(schema.size(), 1);
+
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->predicate_columns = {field_projection(0)};
+    request->non_predicate_columns = {field_projection(0)};
+    request->local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0));
+    request->conjuncts.push_back(
+            VExprContext::create_shared(std::make_shared<NullableInt32GreaterThanExpr>(
+                    0, ConditionCacheContext::GRANULE_SIZE)));
+    ASSERT_TRUE(reader->open(request).ok());
+
+    auto ctx = std::make_shared<ConditionCacheContext>();
+    ctx->is_hit = true;
+    ctx->filter_result =
+            std::make_shared<std::vector<bool>>(std::vector<bool> {false, true, false});
+    reader->set_condition_cache_context(ctx);
+
+    int injection_count = 0;
+    ScopedDebugPoint debug_point(
+            "OrcReader._skip_condition_cache_false_granules.before_seek_to_row", [&]() {
+                ++injection_count;
+                io_ctx->should_stop = true;
+                throw ::orc::ParseError("stop");
+            });
+
+    Block block = build_file_block(schema);
+    size_t rows = 123;
+    bool eof = false;
+    auto status = reader->get_block(&block, &rows, &eof);
+    EXPECT_EQ(injection_count, 1);
+    ASSERT_TRUE(status.ok()) << status;
+    EXPECT_TRUE(eof);
+    EXPECT_EQ(rows, 0);
+    EXPECT_EQ(block.rows(), 0);
+}
+
 TEST_F(NewOrcReaderTest, ConditionCacheHitHandlesSplitWithoutSelectedStripe) {
     const auto multi_stripe_file_path = (_test_dir / "condition_cache_empty_split.orc").string();
     write_multi_stripe_orc_int_file(multi_stripe_file_path);