This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch orc
in repository https://gitbox.apache.org/repos/asf/doris-thirdparty.git
The following commit(s) were added to refs/heads/orc by this push:
new 7955b007759 [fix](orc) Advance past SARG-filtered stripe suffixes when
seeking (#431)
7955b007759 is described below
commit 7955b007759e9c8c5945b8de9fbf1e438cfe60d4
Author: Gabriel <[email protected]>
AuthorDate: Tue Nov 3 16:52:19 2026 +0800
[fix](orc) Advance past SARG-filtered stripe suffixes when seeking (#431)
When seekToRow advances beyond every selected row group of an initialized
stripe, the returned row offset is the stripe-end sentinel. Treating it as
a row-group index can access past RowIndex entries or prematurely stop
reading a partially filled final row group.
Initialize the next selected stripe, respecting the configured stripe range,
before positioning column readers. Add 32 cases covering initialized and
fresh seeks, exact and partial index groups, lazy followers, filtered
intermediate stripes, split boundaries, EOF and backward seeks.
Validation: 56 focused ORC reader and predicate-pushdown tests passed under
AddressSanitizer. All 16 initialized-stripe cases fail without the fix;
all 32 new cases pass with it. clang-format 16 checks passed.
---
c++/src/Reader.cc | 61 ++++++++++++++-----------
c++/test/TestPredicatePushdown.cc | 93 +++++++++++++++++++++++++++++++++++++++
2 files changed, 129 insertions(+), 25 deletions(-)
diff --git a/c++/src/Reader.cc b/c++/src/Reader.cc
index d5f83fc3590..af704655f35 100644
--- a/c++/src/Reader.cc
+++ b/c++/src/Reader.cc
@@ -196,7 +196,8 @@ namespace orc {
selectedColumns[*typeId] = true;
} else {
std::stringstream buffer;
- buffer << "Invalid type id selected " << *typeId << " out of " <<
selectedColumns.size();
+ buffer << "Invalid type id selected " << *typeId << " out of "
+ << selectedColumns.size();
throw ParseError(buffer.str());
}
}
@@ -351,8 +352,7 @@ namespace orc {
while (current != nullptr) {
if (current->getSubtypeCount() == 0) {
current->setReaderCategory(ReaderCategory::FILTER_CHILD);
- } else if (current->getKind() == TypeKind::LIST
- || current->getKind() == TypeKind::MAP) {
+ } else if (current->getKind() == TypeKind::LIST ||
current->getKind() == TypeKind::MAP) {
current->setReaderCategory(ReaderCategory::FILTER_COMPOUND_ELEMENT);
} else {
current->setReaderCategory(ReaderCategory::FILTER_PARENT);
@@ -372,8 +372,7 @@ namespace orc {
if (child->getSubtypeCount() == 0) {
// Leaf node (no children)
child->setReaderCategory(ReaderCategory::FILTER_CHILD);
- } else if (child->getKind() == TypeKind::LIST
- || child->getKind() == TypeKind::MAP) {
+ } else if (child->getKind() == TypeKind::LIST || child->getKind()
== TypeKind::MAP) {
child->setReaderCategory(ReaderCategory::FILTER_COMPOUND_ELEMENT);
// Recursively process its children
processChildren(child);
@@ -410,8 +409,7 @@ namespace orc {
while (current != nullptr) {
if (current->getSubtypeCount() == 0) {
current->setReaderCategory(ReaderCategory::FILTER_CHILD);
- } else if (current->getKind() == TypeKind::LIST
- || current->getKind() == TypeKind::MAP) {
+ } else if (current->getKind() == TypeKind::LIST ||
current->getKind() == TypeKind::MAP) {
current->setReaderCategory(ReaderCategory::FILTER_COMPOUND_ELEMENT);
} else {
current->setReaderCategory(ReaderCategory::FILTER_PARENT);
@@ -431,8 +429,7 @@ namespace orc {
if (child->getSubtypeCount() == 0) {
// Leaf node (no children)
child->setReaderCategory(ReaderCategory::FILTER_CHILD);
- } else if (child->getKind() == TypeKind::LIST
- || child->getKind() == TypeKind::MAP) {
+ } else if (child->getKind() == TypeKind::LIST || child->getKind()
== TypeKind::MAP) {
child->setReaderCategory(ReaderCategory::FILTER_COMPOUND_ELEMENT);
// Recursively process its children
processChildren(child);
@@ -591,6 +588,17 @@ namespace orc {
currentRowInStripe =
advanceToNextRowGroup(currentRowInStripe, rowsInCurrentStripe,
footer->rowindexstride(),
sargsApplier->getNextSkippedRows());
+ // A filtered suffix returns the stripe end, not a valid row-group
index.
+ // For 3000 rows at stride 1000, seeking group 3 would access past the
index.
+ // Initialize the next selected stripe before positioning its column
readers.
+ if (currentRowInStripe == rowsInCurrentStripe) {
+ ++currentStripe;
+ currentRowInStripe = 0;
+ startNextStripe(startReadPhase);
+ if (currentStripe >= lastStripe) {
+ return;
+ }
+ }
}
}
@@ -789,7 +797,8 @@ namespace orc {
return std::unique_ptr<StripeInformation>(new StripeInformationImpl(
stripeInfo.offset(), stripeInfo.indexlength(), stripeInfo.datalength(),
stripeInfo.footerlength(), stripeInfo.numberofrows(),
contents->stream.get(),
- *contents->pool, contents->compression, contents->blockSize,
contents->readerMetrics, nullptr));
+ *contents->pool, contents->compression, contents->blockSize,
contents->readerMetrics,
+ nullptr));
}
FileVersion ReaderImpl::getFormatVersion() const {
@@ -1114,21 +1123,22 @@ namespace orc {
return getMemoryUse(stripeIx, selectedColumns);
}
- void ReaderImpl::getSelectedColumns(const std::list<std::string>& names,
std::vector<bool>& selectedColumns) {
- selectedColumns.clear();
-
selectedColumns.assign(static_cast<size_t>(contents->footer->types_size()),
false);
- ColumnSelector column_selector(contents.get());
- if (contents->schema->getKind() == STRUCT && names.begin() !=
names.end()) {
- for (std::list<std::string>::const_iterator field = names.begin();
field != names.end();
- ++field) {
- column_selector.updateSelectedByName(selectedColumns, *field);
- }
- } else {
- // default is to select all columns
- std::fill(selectedColumns.begin(), selectedColumns.end(), true);
+ void ReaderImpl::getSelectedColumns(const std::list<std::string>& names,
+ std::vector<bool>& selectedColumns) {
+ selectedColumns.clear();
+
selectedColumns.assign(static_cast<size_t>(contents->footer->types_size()),
false);
+ ColumnSelector column_selector(contents.get());
+ if (contents->schema->getKind() == STRUCT && names.begin() != names.end())
{
+ for (std::list<std::string>::const_iterator field = names.begin(); field
!= names.end();
+ ++field) {
+ column_selector.updateSelectedByName(selectedColumns, *field);
}
- column_selector.selectParents(selectedColumns, *contents->schema.get());
- selectedColumns[0] = true; // column 0 is selected by default
+ } else {
+ // default is to select all columns
+ std::fill(selectedColumns.begin(), selectedColumns.end(), true);
+ }
+ column_selector.selectParents(selectedColumns, *contents->schema.get());
+ selectedColumns[0] = true; // column 0 is selected by default
}
uint64_t ReaderImpl::getMemoryUseByName(const std::list<std::string>& names,
int stripeIx) {
@@ -1301,7 +1311,8 @@ namespace orc {
currentStripeInfo.offset(), currentStripeInfo.indexlength(),
currentStripeInfo.datalength(), currentStripeInfo.footerlength(),
currentStripeInfo.numberofrows(), contents->stream.get(),
*contents->pool,
- contents->compression, contents->blockSize, contents->readerMetrics,
¤tStripeFooter));
+ contents->compression, contents->blockSize, contents->readerMetrics,
+ ¤tStripeFooter));
streams.clear();
contents->stream->beforeReadStripe(std::move(currentStripeInformation),
selectedColumns,
streams);
diff --git a/c++/test/TestPredicatePushdown.cc
b/c++/test/TestPredicatePushdown.cc
index e949fc2898d..8922a9da2ce 100644
--- a/c++/test/TestPredicatePushdown.cc
+++ b/c++/test/TestPredicatePushdown.cc
@@ -16,6 +16,8 @@
* limitations under the License.
*/
+#include <tuple>
+
#include "MemoryInputStream.hh"
#include "MemoryOutputStream.hh"
#include "orc/OrcFile.hh"
@@ -26,6 +28,97 @@ namespace orc {
static const int DEFAULT_MEM_STREAM_SIZE = 10 * 1024 * 1024; // 10M
+ class SeekPastFilteredStripeTest
+ : public testing::TestWithParam<std::tuple<uint64_t, uint64_t, bool,
bool, bool>> {};
+
+ class SeekPassThroughFilter : public ORCFilter {
+ public:
+ void filter(ColumnVectorBatch& data, uint16_t* sel, uint16_t size,
+ void* /*arg*/) const override {
+ for (uint16_t row = 0; row < size; ++row) {
+ sel[row] = row;
+ }
+ data.numElements = size;
+ }
+ };
+
+ TEST_P(SeekPastFilteredStripeTest, AdvancesToNextSelectedStripe) {
+ const auto [rowsPerStripe, stripeCount, initializeStripe, restrictRange,
lazy] = GetParam();
+ MemoryOutputStream stream(DEFAULT_MEM_STREAM_SIZE);
+ auto type =
+
std::unique_ptr<Type>(Type::buildTypeFromString("struct<id:bigint,payload:bigint>"));
+ WriterOptions options;
+
options.setStripeSize(1).setCompression(CompressionKind_NONE).setRowIndexStride(1000);
+ auto writer = createWriter(*type, &stream, options);
+ auto batch = writer->createRowBatch(rowsPerStripe);
+ auto& root = dynamic_cast<StructVectorBatch&>(*batch);
+ auto& ids = dynamic_cast<LongVectorBatch&>(*root.fields[0]);
+ auto& payloads = dynamic_cast<LongVectorBatch&>(*root.fields[1]);
+ for (uint64_t stripe = 0; stripe < stripeCount; ++stripe) {
+ for (uint64_t row = 0; row < rowsPerStripe; ++row) {
+ ids.data[row] = static_cast<int64_t>(row + (stripe == 1 ? 10000 : 0));
+ payloads.data[row] = static_cast<int64_t>(stripe * rowsPerStripe +
row);
+ }
+ root.numElements = ids.numElements = payloads.numElements =
rowsPerStripe;
+ writer->add(*batch);
+ }
+ writer->close();
+
+ auto reader = createReader(
+ std::make_unique<MemoryInputStream>(stream.getData(),
stream.getLength()), ReaderOptions());
+ ASSERT_EQ(stripeCount, reader->getNumberOfStripes());
+ ASSERT_EQ(rowsPerStripe, reader->getStripe(0)->getNumberOfRows());
+ RowReaderOptions readOptions;
+ readOptions.searchArgument(SearchArgumentFactory::newBuilder()
+ ->lessThan("id", PredicateDataType::LONG,
Literal(int64_t(1000)))
+ .build());
+ if (restrictRange) {
+ auto stripe = reader->getStripe(0);
+ readOptions.range(stripe->getOffset(), stripe->getLength());
+ }
+ SeekPassThroughFilter filter;
+ if (lazy) {
+ readOptions.filter(std::list<std::string>{"id"});
+ }
+ auto rowReader = reader->createRowReader(readOptions, lazy ? &filter :
nullptr);
+ auto result = rowReader->createRowBatch(500);
+ if (initializeStripe) {
+ ASSERT_TRUE(rowReader->next(*result));
+ ASSERT_EQ(500, result->numElements);
+ }
+
+ // The suffix has no selected row groups. An initialized stripe must take
the same
+ // path as a fresh seek, including skipping the entirely filtered middle
stripe.
+ rowReader->seekToRow(rowsPerStripe - 1);
+ if (stripeCount == 1 || restrictRange) {
+ EXPECT_FALSE(rowReader->next(*result));
+ EXPECT_EQ(0, result->numElements);
+ EXPECT_EQ(restrictRange ? rowsPerStripe : reader->getNumberOfRows(),
+ rowReader->getRowNumber());
+ } else {
+ ASSERT_TRUE(rowReader->next(*result));
+ EXPECT_EQ(rowsPerStripe * 2, rowReader->getRowNumber());
+ ASSERT_EQ(500, result->numElements);
+ auto& values =
+
dynamic_cast<LongVectorBatch&>(*dynamic_cast<StructVectorBatch&>(*result).fields[0]);
+ auto& resultPayloads =
+
dynamic_cast<LongVectorBatch&>(*dynamic_cast<StructVectorBatch&>(*result).fields[1]);
+ for (uint64_t row = 0; row < result->numElements; ++row) {
+ EXPECT_EQ(static_cast<int64_t>(row), values.data[row]);
+ EXPECT_EQ(static_cast<int64_t>(rowsPerStripe * 2 + row),
resultPayloads.data[row]);
+ }
+ }
+
+ rowReader->seekToRow(0);
+ ASSERT_TRUE(rowReader->next(*result));
+ EXPECT_EQ(0, rowReader->getRowNumber());
+ }
+
+ INSTANTIATE_TEST_SUITE_P(FilteredSuffix, SeekPastFilteredStripeTest,
+ testing::Combine(testing::Values(uint64_t(3000),
uint64_t(3500)),
+ testing::Values(uint64_t(1),
uint64_t(3)),
+ testing::Bool(), testing::Bool(),
testing::Bool()));
+
void createMemTestFile(MemoryOutputStream& memStream, uint64_t
rowIndexStride) {
MemoryPool* pool = getDefaultPool();
auto type =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
