This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 896a722a60e branch-4.1: [fix](parquet) Restore column ownership after
interrupted reads #66780 (#68840)
896a722a60e is described below
commit 896a722a60ecd465a605da927a121dba5b9d1586
Author: Mr.Vitvara Varavithya <[email protected]>
AuthorDate: Sat Oct 10 15:33:13 2026 +0700
branch-4.1: [fix](parquet) Restore column ownership after interrupted reads
#66780 (#68840)
### What problem does this PR solve?
Issue Number: None
Related PR: #66780
Problem Summary:
Cherry-pick of #66780 to branch-4.1. When a Parquet scalar column is
read directly into the destination column
(`read_directly_into_dst_logical_column()`), `read_column_data` moves
the caller's only `ColumnPtr` into `resolved_column`; if any read step
returns before `convert()` moves it back (for example when the scan is
stopped), the caller is left with a null column and the BE crashes. The
fix restores ownership with a `Defer` on every early return, exactly as
on master.
Conflict resolution: on branch-4.1 `read_column_data` ends with `return
_converter->convert(...)` directly, so the last hunk becomes `return
convert_column();` (master has `RETURN_IF_ERROR(convert_column()); …
return Status::OK();`) — same behaviour.
The unit test from #66780 is carried over as a standalone file,
`parquet_scalar_column_reader_test.cpp`: the file it extends on master
(`parquet_column_chunk_reader_test.cpp`) and the lazy-dictionary
fixtures it builds on (#66000) are not on branch-4.1, so the test brings
the minimal in-memory column-chunk fixture it needs.
### Release note
Fix a BE crash when a Parquet read into a nested scalar column is
interrupted before the column is converted.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test:
`ParquetScalarColumnReaderTest.NestedReadRestoresColumnWhenStopped`
(`be/test/format/parquet/parquet_scalar_column_reader_test.cpp`)
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [x] No.
- [ ] Yes.
- 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
---
be/src/format/parquet/vparquet_column_reader.cpp | 22 ++-
.../parquet/parquet_scalar_column_reader_test.cpp | 170 +++++++++++++++++++++
2 files changed, 188 insertions(+), 4 deletions(-)
diff --git a/be/src/format/parquet/vparquet_column_reader.cpp
b/be/src/format/parquet/vparquet_column_reader.cpp
index 82a0b81bc85..7c9d57d2e14 100644
--- a/be/src/format/parquet/vparquet_column_reader.cpp
+++ b/be/src/format/parquet/vparquet_column_reader.cpp
@@ -41,6 +41,7 @@
#include "format/table/iceberg_default_value.h"
#include "io/fs/tracing_file_reader.h"
#include "runtime/runtime_profile.h"
+#include "util/defer_op.h"
namespace doris {
#include "common/compile_check_begin.h"
@@ -548,11 +549,26 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
ColumnPtr resolved_column =
_converter->get_physical_column(_field_schema->physical_type,
_field_schema->data_type,
doris_column, type,
is_dict_filter);
+ // Direct reads transfer the caller's only ColumnPtr so mutate() can avoid
cloning it. Restore
+ // that ownership if any read step returns before convert() moves the
column back.
+ bool restore_doris_column = false;
+ Defer restore_column([&]() {
+ if (restore_doris_column) {
+ doris_column = std::move(resolved_column);
+ }
+ });
if (_converter->read_directly_into_dst_logical_column()) {
DCHECK_EQ(resolved_column.get(), doris_column.get());
resolved_column = std::move(doris_column);
+ restore_doris_column = true;
}
DataTypePtr& resolved_type = _converter->get_physical_type();
+ auto convert_column = [&]() -> Status {
+ RETURN_IF_ERROR(_converter->convert(resolved_column,
_field_schema->data_type, type,
+ doris_column, is_dict_filter));
+ restore_doris_column = false;
+ return Status::OK();
+ };
_def_levels.clear();
_rep_levels.clear();
@@ -561,8 +577,7 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
if (_in_nested) {
RETURN_IF_ERROR(_read_nested_column(resolved_column, resolved_type,
filter_map, batch_size,
read_rows, eof, is_dict_filter));
- return _converter->convert(resolved_column, _field_schema->data_type,
type, doris_column,
- is_dict_filter);
+ return convert_column();
}
int64_t right_row = 0;
@@ -638,8 +653,7 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
}
}
- return _converter->convert(resolved_column, _field_schema->data_type,
type, doris_column,
- is_dict_filter);
+ return convert_column();
}
Status ArrayColumnReader::init(std::unique_ptr<ParquetColumnReader>
element_reader,
diff --git a/be/test/format/parquet/parquet_scalar_column_reader_test.cpp
b/be/test/format/parquet/parquet_scalar_column_reader_test.cpp
new file mode 100644
index 00000000000..98e1d43999f
--- /dev/null
+++ b/be/test/format/parquet/parquet_scalar_column_reader_test.cpp
@@ -0,0 +1,170 @@
+// 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 <gtest/gtest.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "common/cast_set.h"
+#include "common/status.h"
+#include "core/column/column_string.h"
+#include "core/data_type/data_type_string.h"
+#include "format/parquet/parquet_common.h"
+#include "format/parquet/schema_desc.h"
+#include "format/parquet/vparquet_column_reader.h"
+#include "io/fs/file_reader.h"
+#include "io/io_common.h"
+#include "runtime/runtime_state.h"
+#include "util/thrift_util.h"
+
+namespace doris {
+namespace {
+
+// Serves an in-memory byte buffer as the parquet file behind a column reader.
+class InMemoryFileReader final : public io::FileReader {
+public:
+ explicit InMemoryFileReader(std::vector<uint8_t> data) :
_data(std::move(data)) {}
+
+ Status close() override {
+ _closed = true;
+ return Status::OK();
+ }
+
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return _data.size(); }
+ bool closed() const override { return _closed; }
+ int64_t mtime() const override { return 0; }
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ if (offset > _data.size()) {
+ return Status::IOError("Out of bounds");
+ }
+ *bytes_read = std::min(result.size, _data.size() - offset);
+ memcpy(result.data, _data.data() + offset, *bytes_read);
+ return Status::OK();
+ }
+
+private:
+ std::vector<uint8_t> _data;
+ io::Path _path = "parquet_scalar_column_reader_test";
+ bool _closed = false;
+};
+
+struct ColumnChunkFixture {
+ std::vector<uint8_t> data;
+ tparquet::ColumnChunk chunk;
+ FieldSchema field_schema;
+};
+
+Status append_page(tparquet::PageHeader* header, const std::vector<uint8_t>&
payload,
+ std::vector<uint8_t>& data, int64_t* page_offset, int32_t*
page_size) {
+ std::vector<uint8_t> header_bytes;
+ ThriftSerializer serializer(/*compact=*/true, /*initial_buffer_size=*/256);
+ RETURN_IF_ERROR(serializer.serialize(header, &header_bytes));
+
+ *page_offset = data.size();
+ data.insert(data.end(), header_bytes.begin(), header_bytes.end());
+ data.insert(data.end(), payload.begin(), payload.end());
+ *page_size = cast_set<int32_t>(header_bytes.size() + payload.size());
+ return Status::OK();
+}
+
+tparquet::PageHeader make_data_page_header(tparquet::Encoding::type encoding) {
+ tparquet::DataPageHeader data_header;
+ data_header.__set_num_values(1);
+ data_header.__set_encoding(encoding);
+ data_header.__set_definition_level_encoding(tparquet::Encoding::RLE);
+ data_header.__set_repetition_level_encoding(tparquet::Encoding::RLE);
+
+ tparquet::PageHeader header;
+ header.type = tparquet::PageType::DATA_PAGE;
+ header.__set_compressed_page_size(1);
+ header.__set_uncompressed_page_size(1);
+ header.__set_data_page_header(data_header);
+ return header;
+}
+
+// One uncompressed BYTE_ARRAY column chunk holding a single plain-encoded
data page.
+Status make_plain_fixture(ColumnChunkFixture* fixture) {
+ constexpr size_t PREFIX_SIZE = 16;
+ fixture->data.resize(PREFIX_SIZE, 0);
+
+ tparquet::PageHeader header =
make_data_page_header(tparquet::Encoding::PLAIN);
+ int64_t data_page_offset = 0;
+ int32_t data_page_size = 0;
+ RETURN_IF_ERROR(append_page(&header, {0}, fixture->data,
&data_page_offset, &data_page_size));
+
+ auto& metadata = fixture->chunk.meta_data;
+ metadata.__set_type(tparquet::Type::BYTE_ARRAY);
+ metadata.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
+ metadata.__set_num_values(1);
+ metadata.__set_data_page_offset(data_page_offset);
+ metadata.__set_total_compressed_size(
+ cast_set<int64_t>(fixture->data.size() - data_page_offset));
+
+ fixture->field_schema.physical_type = tparquet::Type::BYTE_ARRAY;
+ return Status::OK();
+}
+
+} // namespace
+
+// A direct read moves the caller's column into the reader; a read that stops
before convert()
+// hands it back, so the caller still holds a column instead of a null pointer.
+TEST(ParquetScalarColumnReaderTest, NestedReadRestoresColumnWhenStopped) {
+ ColumnChunkFixture fixture;
+ ASSERT_TRUE(make_plain_fixture(&fixture).ok());
+ auto file_reader =
std::make_shared<InMemoryFileReader>(std::move(fixture.data));
+ DataTypePtr string_type = std::make_shared<DataTypeString>();
+ fixture.field_schema.data_type = string_type;
+ fixture.field_schema.parquet_schema.__set_type(tparquet::Type::BYTE_ARRAY);
+
+ RowRanges row_ranges;
+ row_ranges.add({0, 1});
+ io::IOContext io_ctx;
+ ScalarColumnReader<true, false> reader(row_ranges, 1, fixture.chunk,
nullptr, nullptr, &io_ctx);
+ reader.set_column_in_nested();
+
+ TQueryOptions query_options;
+ query_options.__set_enable_parquet_file_page_cache(false);
+ RuntimeState runtime_state(query_options, TQueryGlobals());
+ ASSERT_TRUE(reader.init(file_reader, &fixture.field_schema,
+ /*max_buf_size=*/1024 * 1024, &runtime_state)
+ .ok());
+
+ ColumnPtr column = ColumnString::create();
+ FilterMap filter_map;
+ ASSERT_TRUE(filter_map.init(nullptr, 0, false).ok());
+ size_t read_rows = 0;
+ bool eof = false;
+ io_ctx.should_stop = true;
+
+ Status status = reader.read_column_data(column, string_type, nullptr,
filter_map, 1, &read_rows,
+ &eof, false);
+ EXPECT_TRUE(status.is<ErrorCode::END_OF_FILE>()) << status;
+ ASSERT_TRUE(column);
+ EXPECT_TRUE(column->empty());
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
