[fix](be) Fix ORC condition-cache seeks past filtered stripe suffixes (#70079)

### What problem does this PR solve?

Related PR: https://github.com/apache/doris-thirdparty/pull/431 (merge
the ORC dependency fix first)

A condition-cache granule can start in a stripe's SARG-filtered suffix
while containing a matching row in the next stripe. On an initialized
stripe, ORC's `seekToRow()` uses the stripe-end sentinel as a row-group
index. This can access past the index when the stripe ends on an index
boundary, or prematurely return EOF when the last row group is partial.

The same-stripe fast path originates in [Apache ORC
ORC-1170](https://github.com/apache/orc/pull/1114); the same missing
stripe-end handling remains in the current upstream C++ reader.

Pin the ORC fix that initializes the next selected stripe before
positioning its column readers. Preserve SARG pruning, lazy reads and
split boundaries.

Add a Doris reader regression and a 1.5 KB synthetic ORC fixture with
its generator. The fixture has two 20,000-row stripes, a 10,000-row
index stride, and one matching row at the start of each stripe. A cache
hit seeks to row 18,432 in the first stripe's filtered suffix. The SQL
regression compares cache-disabled and cached results with lazy reads
and min/max pruning enabled or disabled, and checks the profile for
actual cache hits and skipped rows.

### Release note

Fix potential crashes and missing rows when ORC condition-cache seeks
reach a SARG-filtered stripe suffix.

### Check List (For Author)

- Test:
- [x] Unit Test: 247 ORC-related BE tests passed under AddressSanitizer,
including the new Doris condition-cache case and 32 dependency boundary
cases.
- [x] Unit Test: 56 focused standalone ORC tests passed under
AddressSanitizer. Reverting the fix makes all 16 initialized-stripe
boundary cases fail.
- [ ] Regression test: SQL suite added; Groovy compilation checked. The
external SQL suite has not been executed locally.
  - [x] clang-format 16 and BE build-hygiene checks passed.
- [ ] Local full-file clang-tidy analysis did not complete within ten
minutes; no completed result is available.
- Behavior changed:
  - [x] Yes. Safely advance to the next matching stripe or EOF.
- Does this need documentation?
  - [x] No.

be/test/format_v2/orc/orc_reader_test.cpp

@@ -6343,6 +6343,72 @@ TEST_F(NewOrcReaderTest, ConditionCacheHitSkipsFalseGranulesBeforeColumnRead) {
     EXPECT_EQ(rows, 0);
 }
 
+TEST_F(NewOrcReaderTest, ConditionCacheHitSeeksPastSargFilteredStripeSuffix) {
+    const auto file_path = find_repo_file(
+            "regression-test/data/external_table_p0/tvf/test_data/condition_cache_seek.orc");
+    ASSERT_FALSE(file_path.empty());
+    for (const bool lazy : {false, true}) {
+        for (const bool min_max : {false, true}) {
+            SCOPED_TRACE(fmt::format("lazy={}, min_max={}", lazy, min_max));
+            auto io_ctx = std::make_shared<io::IOContext>();
+            auto reader = create_reader_for_path(file_path.string(), nullptr, io_ctx);
+            TQueryOptions options;
+            options.__set_enable_orc_lazy_mat(lazy);
+            options.__set_enable_orc_filter_by_min_max(min_max);
+            RuntimeState state {options, TQueryGlobals()};
+            ASSERT_TRUE(reader->init(&state).ok());
+            reader->set_batch_size(ConditionCacheContext::GRANULE_SIZE);
+
+            std::vector<format::ColumnDefinition> schema;
+            ASSERT_TRUE(reader->get_schema(&schema).ok());
+            auto request = std::make_shared<format::FileScanRequest>();
+            request->predicate_columns = {field_projection(0)};
+            request->non_predicate_columns = {field_projection(1)};
+            request->local_positions = {{format::LocalColumnId(0), format::LocalIndex(0)},
+                                        {format::LocalColumnId(1), format::LocalIndex(1)}};
+            request->conjuncts.push_back(VExprContext::create_shared(
+                    std::make_shared<NullableInt32GreaterThanExpr>(0, 5000)));
+            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>>(20, false);
+            (*ctx->filter_result)[0] = true;
+            (*ctx->filter_result)[9] = true;
+            reader->set_condition_cache_context(ctx);
+            int seeks = 0;
+            ScopedDebugPoint debug_point(
+                    "OrcReader._skip_condition_cache_false_granules.before_seek_to_row",
+                    [&]() { ++seeks; });
+
+            // Granule 9 contains the next stripe's matching row, but starts in the
+            // current stripe's filtered suffix. Both columns must resume in stripe 2.
+            std::vector<int32_t> ids;
+            std::vector<int64_t> payloads;
+            bool eof = false;
+            while (!eof) {
+                Block block = build_file_block(schema);
+                size_t rows = 0;
+                ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+                const auto& id_column = assert_cast<const ColumnInt32&>(
+                        assert_cast<const ColumnNullable&>(*block.get_by_position(0).column)
+                                .get_nested_column());
+                const auto& payload_column = assert_cast<const ColumnInt64&>(
+                        assert_cast<const ColumnNullable&>(*block.get_by_position(1).column)
+                                .get_nested_column());
+                for (size_t row = 0; row < rows; ++row) {
+                    ids.push_back(id_column.get_element(row));
+                    payloads.push_back(payload_column.get_element(row));
+                }
+            }
+            EXPECT_EQ(ids, (std::vector<int32_t> {10000, 10001}));
+            EXPECT_EQ(payloads, (std::vector<int64_t> {0, 20000}));
+            EXPECT_GT(seeks, 0);
+            EXPECT_GT(io_ctx->condition_cache_filtered_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();

regression-test/data/external_table_p0/tvf/test_data/GenerateOrcConditionCacheSeek.cpp

@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <memory>
+#include <orc/OrcFile.hh>
+#include <orc/Writer.hh>
+
+// Link against the ORC library used by Doris and pass the output fixture path.
+// Each stripe has 20,000 rows and two 10,000-row index groups. Only its first row
+// matches id > 5000. The second match makes cache granule [18432, 20480) true,
+// so a cache hit seeks into the first stripe's SARG-filtered suffix at row 18432.
+int main(int argc, char** argv) {
+    if (argc != 2) {
+        return 1;
+    }
+    auto stream = orc::writeLocalFile(argv[1]);
+    auto type = std::unique_ptr<orc::Type>(
+            orc::Type::buildTypeFromString("struct<id:int,payload:bigint>"));
+    orc::WriterOptions options;
+    options.setStripeSize(1).setRowIndexStride(10000).setCompression(orc::CompressionKind_NONE);
+    auto writer = orc::createWriter(*type, stream.get(), options);
+    auto batch = writer->createRowBatch(20000);
+    auto& root = dynamic_cast<orc::StructVectorBatch&>(*batch);
+    auto& ids = dynamic_cast<orc::LongVectorBatch&>(*root.fields[0]);
+    auto& payloads = dynamic_cast<orc::LongVectorBatch&>(*root.fields[1]);
+    for (int64_t stripe = 0; stripe < 2; ++stripe) {
+        for (int64_t row = 0; row < 20000; ++row) {
+            ids.data[row] = row == 0 ? 10000 + stripe : 0;
+            payloads.data[row] = stripe * 20000 + row;
+        }
+        root.numElements = ids.numElements = payloads.numElements = 20000;
+        writer->add(*batch);
+    }
+    writer->close();
+    stream->close();
+    return 0;
+}

regression-test/suites/external_table_p0/tvf/test_orc_condition_cache_seek.groovy

@@ -0,0 +1,110 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import com.azure.storage.blob.BlobClientBuilder
+import com.azure.storage.common.StorageSharedKeyCredential
+import org.apache.doris.regression.action.ProfileAction
+
+suite("test_orc_condition_cache_seek", "p0,external") {
+    String provider = getS3Provider()
+    String endpoint = getS3Endpoint()
+    if (provider.equalsIgnoreCase("AZURE")) {
+        if (endpoint == "azure-endpoint") {
+            endpoint = "https://${getS3AK()}.blob.core.windows.net"
+        } else if (!endpoint.contains("://")) {
+            endpoint = "https://${endpoint}"
+        }
+    }
+    String bucket = context.config.otherConfigs.get("s3BucketName")
+    String key = "orc_condition_cache_seek/${UUID.randomUUID()}/values.orc"
+    File fixture = new File(context.dataPath, "test_data/condition_cache_seek.orc")
+    def blobClient = null
+    def s3Client = null
+    boolean uploaded = false
+    Map<String, String> originalSettings = [:]
+    for (String variable : ["enable_condition_cache", "enable_orc_lazy_materialization",
+                            "enable_orc_filter_by_min_max", "batch_size", "enable_sql_cache",
+                            "enable_query_cache", "enable_profile", "profile_level", "parallel_pipeline_task_num",
+                            "max_file_scanners_concurrency"]) {
+        originalSettings[variable] = sql("SHOW VARIABLES LIKE '${variable}'")[0][1].toString()
+    }
+    try {
+        if (provider.equalsIgnoreCase("AZURE")) {
+            blobClient = new BlobClientBuilder().endpoint(endpoint).containerName(bucket)
+                    .blobName(key).credential(new StorageSharedKeyCredential(getS3AK(), getS3SK()))
+                    .buildClient()
+            blobClient.uploadFromFile(fixture.absolutePath)
+        } else {
+            s3Client = getS3Client()
+            s3Client.putObject(bucket, key, fixture)
+        }
+        uploaded = true
+        String source = """S3('uri'='s3://${bucket}/${key}', 'format'='orc',
+                'provider'='${provider}', 's3.endpoint'='${endpoint}', 's3.region'='${getS3Region()}',
+                's3.access_key'='${getS3AK()}', 's3.secret_key'='${getS3SK()}')"""
+        sql "SET enable_sql_cache = false"
+        sql "SET enable_query_cache = false"
+        sql "SET enable_profile = true"
+        sql "SET profile_level = 2"
+        sql "SET parallel_pipeline_task_num = 1"
+        sql "SET max_file_scanners_concurrency = 1"
+        sql "SET batch_size = 2048"
+        // A full LRU-K shard needs one scan for ghost history and another for admission.
+        // Allow a third visit to one backend even when scans are spread across all backends.
+        int attempts = 2 * sql("SHOW BACKENDS").size() + 1
+        for (boolean minMax : [false, true]) {
+            sql "SET enable_orc_filter_by_min_max = ${minMax}"
+            for (boolean lazy : [false, true]) {
+                sql "SET enable_orc_lazy_materialization = ${lazy}"
+                sql "SET enable_condition_cache = false"
+                quickTest("uncached", "SELECT id, payload FROM ${source} WHERE id > 5000 ORDER BY id",
+                        "SELECT 10000, CAST(0 AS BIGINT) UNION ALL SELECT 10001, CAST(20000 AS BIGINT) ORDER BY 1")
+                sql "SET enable_condition_cache = true"
+                boolean cacheHit = false
+                for (int attempt = 0; attempt < attempts && !cacheHit; ++attempt) {
+                    String token = "orc_seek_${UUID.randomUUID().toString().replace('-', '')}"
+                    // The first match populates granule 0. The next stripe's match populates
+                    // granule 9, whose start (18432) lies in the first stripe's pruned suffix.
+                    quickTest("cached",
+                            "SELECT id, payload FROM (SELECT id, payload, '${token}' AS marker " +
+                                    "FROM ${source} WHERE id > 5000) t ORDER BY id",
+                            "SELECT 10000, CAST(0 AS BIGINT) UNION ALL " +
+                                    "SELECT 10001, CAST(20000 AS BIGINT) ORDER BY 1")
+                    String profile = new ProfileAction(context).getProfileBySql(token,
+                            ["ConditionCacheHit", "ConditionCacheFilteredRows"])
+                    cacheHit = (profile =~ /(?m)ConditionCacheHit:\s*[1-9]/).find()
+                    if (cacheHit) {
+                        assertTrue((profile =~ /(?m)ConditionCacheFilteredRows:\s*[1-9]/).find(), profile)
+                    }
+                }
+                assertTrue(cacheHit, "No condition cache hit after ${attempts} scans")
+            }
+        }
+    } finally {
+        try {
+            originalSettings.each { variable, value -> sql "SET ${variable} = ${value}" }
+        } finally {
+            if (uploaded) {
+                if (blobClient != null) {
+                    blobClient.delete()
+                } else {
+                    s3Client.deleteObject(bucket, key)
+                }
+            }
+        }
+    }
+}