diff --git a/be/src/format_v2/orc/orc_reader.cpp b/be/src/format_v2/orc/orc_reader.cpp index e7509ebd078364..83cb929e18bccc 100644 --- a/be/src/format_v2/orc/orc_reader.cpp +++ b/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; diff --git a/be/test/format_v2/orc/orc_reader_test.cpp b/be/test/format_v2/orc/orc_reader_test.cpp index 2663f3fea07a7f..1058d774d44e60 100644 --- a/be/test/format_v2/orc/orc_reader_test.cpp +++ b/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(); + auto reader = create_reader_for_path(file_path, nullptr, io_ctx); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + ASSERT_EQ(schema.size(), 1); + + auto request = std::make_shared(); + 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( + 0, ConditionCacheContext::GRANULE_SIZE))); + ASSERT_TRUE(reader->open(request).ok()); + + auto ctx = std::make_shared(); + ctx->is_hit = true; + ctx->filter_result = + std::make_shared>(std::vector {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);