github-actions[bot] commented on code in PR #65931:
URL: https://github.com/apache/doris/pull/65931#discussion_r3635726315
##########
be/src/format_v2/parquet/native_schema_node.cpp:
##########
@@ -68,24 +68,39 @@ Status build_native_schema_node(const DataTypePtr&
projected_type,
return Status::Corruption("Parquet column {} is not a STRUCT",
file_schema.name);
}
const auto* struct_type = assert_cast<const
DataTypeStruct*>(type.get());
- std::map<std::string, const ParquetColumnSchema*> file_children;
- for (const auto& child : file_schema.children) {
- file_children.emplace(to_lower(child->name), child.get());
- }
auto node = std::make_shared<NativeStructSchemaNode>();
for (size_t i = 0; i < struct_type->get_elements().size(); ++i) {
const auto& table_name = struct_type->get_element_name(i);
- // Native metadata keeps writer casing. Match normalized names
while preserving the
- // original file name used to address the physical child reader.
- const auto child_it = file_children.find(to_lower(table_name));
- if (child_it == file_children.end()) {
+ const ParquetColumnSchema* file_child = nullptr;
+ for (const auto& child : file_schema.children) {
Review Comment:
[P2] Keep STRUCT child resolution indexed
This exact-name scan restarts from the first physical child for every
projected field, so a full N-field STRUCT now performs roughly N(N+1)/2
comparisons; fallback scans again and recomputes lowercase strings inside the
loop. `NativeColumnReader::create()` rebuilds this mapping for predicate/output
readers in every selected Row Group, making wide nested schemas pay the
quadratic cost repeatedly. This is distinct from the existing case-collision
correctness thread. Build exact-name and normalized unique/ambiguous indices
once per physical STRUCT, then preserve the current exact-first semantics
through indexed lookups.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -86,6 +86,82 @@ Status validate_uncompressed_page_sizes(const
tparquet::PageHeader& header,
return Status::OK();
}
+Status validate_fixed_width_page_size(const tparquet::PageHeader& header,
int32_t type_length,
+ level_t max_rep_level, level_t
max_def_level,
+ bool schema_is_required) {
+ if (type_length <= 0 || max_rep_level != 0) {
Review Comment:
[P1] Close the remaining compressed data-page preallocation paths
This still returns before validation for repeated V2 leaves, although V2
provides `num_values - num_nulls` and exact repetition/definition byte lengths.
The `type_length <= 0` and V1 optional fallbacks likewise reach
`_reserve_decompress_buf(uncompressed_size)` before Snappy reads the frame's
true decoded length. A tiny compressed `ARRAY<INT32>` V2 page or one-value
`BYTE_ARRAY`/optional V1 page can therefore advertise `INT32_MAX` and make
Doris reserve nearly 2 GiB before rejecting it. The existing thread covered
flat optional V2; these are distinct remaining layouts. Apply the exact V2
formula regardless of repetition, preflight codec output before reserving for
layouts without a statically known extent, and add compressed regressions for
both paths.
##########
be/test/format_v2/table/remote_doris_reader_test.cpp:
##########
@@ -467,4 +600,140 @@ TEST(RemoteDorisV2ReaderTest,
RejectsInvalidRemoteDorisRange) {
EXPECT_FALSE(reader->init(&state).ok());
}
+TEST(RemoteDorisV2ReaderTest, RuntimeCancellationInterruptsBlockedFlightDoGet)
{
+ BlockingFlightServer server(BlockingFlightServer::Mode::DO_GET);
+ const auto server_status = server.start();
+ ASSERT_TRUE(server_status.ok()) << server_status;
+ ObjectPool pool;
+ DescriptorTbl* desc_tbl = nullptr;
+ const auto slots = remote_slots(&pool, &desc_tbl);
+ RuntimeState state;
+ RuntimeProfile profile("remote_doris_v2_blocked_doget_test");
+ auto io_ctx = std::make_shared<io::IOContext>();
+ auto reader = create_flight_reader(&profile, remote_doris_range(server),
slots, io_ctx);
+ ASSERT_TRUE(reader->init(&state).ok());
+ auto request = std::make_shared<FileScanRequest>();
+ FileScanRequestBuilder builder(request.get());
+ ASSERT_TRUE(builder.add_non_predicate_column(LocalColumnId(0)).ok());
+
+ auto open_result =
+ std::async(std::launch::async, [&] { return
reader->open(std::move(request)); });
+ const bool entered = server.wait_until_entered(std::chrono::seconds(2));
+ if (!entered &&
+ open_result.wait_for(std::chrono::milliseconds(0)) ==
std::future_status::ready) {
+ FAIL() << "Flight DoGet failed before reaching the server: " <<
open_result.get();
+ }
+ ASSERT_TRUE(entered);
+ state.cancel(Status::Cancelled("cancel blocked Flight DoGet"));
+ const bool interrupted =
+ open_result.wait_for(std::chrono::milliseconds(750)) ==
std::future_status::ready;
+ if (!interrupted) {
+ server.release();
+ }
+ EXPECT_TRUE(interrupted);
+ EXPECT_FALSE(open_result.get().ok());
+}
+
+TEST(RemoteDorisV2ReaderTest,
BlockedDoGetRetainsQueryResourcesUntilWorkerExits) {
+ BlockingFlightServer server(BlockingFlightServer::Mode::DO_GET);
+ const auto server_status = server.start();
+ ASSERT_TRUE(server_status.ok()) << server_status;
+ ObjectPool pool;
+ DescriptorTbl* desc_tbl = nullptr;
+ const auto slots = remote_slots(&pool, &desc_tbl);
+ auto state = std::make_unique<MockRuntimeState>();
+ std::weak_ptr<ResourceContext> resource_ctx =
state->get_query_ctx()->resource_ctx();
+ RuntimeProfile profile("remote_doris_v2_doget_resource_context_test");
+ auto reader = create_flight_reader(&profile, remote_doris_range(server),
slots,
+ std::make_shared<io::IOContext>());
+ ASSERT_TRUE(reader->init(state.get()).ok());
+ auto request = std::make_shared<FileScanRequest>();
+ FileScanRequestBuilder builder(request.get());
+ ASSERT_TRUE(builder.add_non_predicate_column(LocalColumnId(0)).ok());
+
+ auto open_result =
+ std::async(std::launch::async, [&] { return
reader->open(std::move(request)); });
+ ASSERT_TRUE(server.wait_until_entered(std::chrono::seconds(2)));
+ state->cancel(Status::Cancelled("cancel blocked Flight DoGet"));
+ ASSERT_EQ(open_result.wait_for(std::chrono::milliseconds(750)),
std::future_status::ready);
Review Comment:
[P2] Release the server before a fatal async assertion can return
If this 750 ms assertion fails, GoogleTest returns from the test and
destroys `open_result` before `server` (reverse declaration order). A
`std::future` from `std::async(std::launch::async, ...)` waits for its
unfinished task, while the server destructor that would release the blocked
`DoGet` cannot run; the only remaining bound is the default 3600-second
query/Flight deadline. The fatal assertions at lines 626 and 698 have the same
cleanup ordering. Install a scope guard that releases the server before the
future is destroyed, or explicitly release/join before reporting each failure.
##########
be/src/format_v2/parquet/reader/native/column_reader.cpp:
##########
@@ -1347,11 +1354,16 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_data(
}
const DataTypePtr serde_type = is_dict_filter ? file_type :
materialization_type;
- if (_serde_type != serde_type.get() || _dictionary_index_only !=
is_dict_filter) {
+ const bool serde_type_changed = _serde_type != serde_type.get();
Review Comment:
[P2] Compare logical SerDe types before dropping the retained dictionary
With TIMESTAMPTZ mapping enabled, native schema parsing already creates a
`DataTypeTimeStampTz`; `build_parquet_column_schema()` shares that pointer, but
`apply_timestamp_tz_mapping()` replaces it with a newly allocated, semantically
equal type. The pruning probe publishes the projected pointer, then every
dictionary-ID batch selects the native field pointer here and resets
`_materialization_state`; matched-value materialization switches back and
rebuilds the full typed dictionary, repeating each batch. The direct test
passes one shared type pointer and misses this production path. Compare
semantic type/scale (or avoid remapping an already equal type) and add a
multi-batch TIMESTAMPTZ dictionary-filter regression.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -86,6 +86,82 @@ Status validate_uncompressed_page_sizes(const
tparquet::PageHeader& header,
return Status::OK();
}
+Status validate_fixed_width_page_size(const tparquet::PageHeader& header,
int32_t type_length,
+ level_t max_rep_level, level_t
max_def_level,
+ bool schema_is_required) {
+ if (type_length <= 0 || max_rep_level != 0) {
+ return Status::OK();
+ }
+ const bool is_v2 = header.__isset.data_page_header_v2;
+ if (!is_v2 && !header.__isset.data_page_header) {
+ return Status::OK();
+ }
+ const auto encoding =
+ is_v2 ? header.data_page_header_v2.encoding :
header.data_page_header.encoding;
+ if (encoding != tparquet::Encoding::PLAIN &&
+ encoding != tparquet::Encoding::BYTE_STREAM_SPLIT) {
+ return Status::OK();
+ }
+ int32_t num_physical_values = 0;
+ int64_t level_bytes = 0;
+ if (is_v2) {
+ const auto& page = header.data_page_header_v2;
+ if (UNLIKELY(page.num_values < 0 || page.num_nulls < 0 ||
+ page.num_nulls > page.num_values ||
page.repetition_levels_byte_length < 0 ||
+ page.definition_levels_byte_length < 0)) {
+ return Status::Corruption("Parquet data page v2 has invalid value
or level counts");
+ }
+ num_physical_values = page.num_values - page.num_nulls;
+ level_bytes = static_cast<int64_t>(page.repetition_levels_byte_length)
+
+ page.definition_levels_byte_length;
+ } else {
+ if (max_def_level != 0 || !schema_is_required) {
+ return Status::OK();
+ }
+ num_physical_values = header.data_page_header.num_values;
+ }
+ if (level_bytes > std::numeric_limits<int32_t>::max() ||
num_physical_values < 0 ||
+ static_cast<uint64_t>(num_physical_values) >
+ (static_cast<uint64_t>(std::numeric_limits<int32_t>::max()) -
level_bytes) /
+ static_cast<uint32_t>(type_length)) {
+ return Status::Corruption("Parquet fixed-width PLAIN page byte size
overflows");
+ }
+ const int64_t expected = level_bytes +
static_cast<int64_t>(num_physical_values) * type_length;
+ if (UNLIKELY(header.uncompressed_page_size != expected)) {
+ // V2 exposes null and level extents separately, so fixed-width
payload size is known before
+ // decompression even for optional columns and must gate
attacker-controlled allocation.
+ return Status::Corruption("Parquet fixed-width page has {}
uncompressed bytes, expected {}",
+ header.uncompressed_page_size, expected);
+ }
+ return Status::OK();
+}
+
+Status validate_dictionary_page_size(const tparquet::PageHeader& header,
int32_t type_length) {
+ DORIS_CHECK(header.__isset.dictionary_page_header);
+ const int32_t num_values = header.dictionary_page_header.num_values;
+ if (UNLIKELY(num_values < 0 || (num_values == 0 &&
header.uncompressed_page_size != 0))) {
+ // An empty dictionary owns no payload; validate before allocating
from its untrusted size.
+ return Status::Corruption("Parquet dictionary has {} values and {}
uncompressed bytes",
+ num_values, header.uncompressed_page_size);
+ }
+ if (type_length > 0) {
Review Comment:
[P1] Validate variable-width dictionary output before allocating it
`BYTE_ARRAY` keeps `type_length == -1`, so this helper accepts any positive
`uncompressed_page_size`; `_decode_dict_page()` then allocates that advertised
size before reading either the cached or file payload. A one-entry Snappy
dictionary that really expands to a short length-prefixed value can therefore
advertise `INT32_MAX` and force a near-2-GiB allocation before
`GetUncompressedLength` exposes the mismatch. This is distinct from the
existing fixed-width dictionary thread. Preflight the compressed frame's
decoded length before allocation (and require it to match the page header),
with a compressed BYTE_ARRAY regression.
##########
be/src/format_v2/table/remote_doris_reader.cpp:
##########
@@ -99,14 +207,65 @@ class FlightRemoteDorisStream final : public
RemoteDorisStream {
}
private:
+ bool _is_cancelled() const {
+ return (_runtime_state != nullptr && _runtime_state->is_cancelled()) ||
+ (_io_ctx != nullptr && _io_ctx->should_stop);
Review Comment:
[P1] Synchronize the scanner-stop signal read by this watcher
`ScannerContext::stop_scanners()` can call `FileScannerV2::try_stop()` on
the operator thread, which writes `IOContext::should_stop`, while this new
`std::jthread` reads the same field here. `should_stop` is a plain `bool` and
no common lock protects it, so the LIMIT cancellation path has a C++ data race;
the watcher may miss the stop and leave `Next()` blocked until the Flight
deadline. The new blocked-Next test performs the same unsynchronized write.
This is distinct from the earlier missing-propagation thread: make the
lifecycle signal atomic/acquire-release or use a dedicated synchronized stop
token, and exercise that production mechanism in the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
