This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 32670407a51 [fix](parquet) Include file and Variant field context in
footer errors (#69094)
32670407a51 is described below
commit 32670407a51f120b80c1d39563e4bcb6337e87b1
Author: Gabriel <[email protected]>
AuthorDate: Thu Oct 15 14:26:07 2026 +0800
[fix](parquet) Include file and Variant field context in footer errors
(#69094)
### What problem does this PR solve?
Related PR: #69092
Parquet V2 footer/schema errors can reach the client without identifying
the input file. Invalid Variant fallback errors report only a wrapper
name such as `element`, making nested failures difficult to locate.
Include the source file path in native footer errors while preserving
the original status code and message. Invalid Variant fallback errors
also include the Variant-relative field path, declared physical type,
repetition, and child count. Diagnostic paths are constructed only on
failure; normalized repeated primitives report their original
declarations.
This ports #69092 to master, retaining master's metadata ownership and
`preserve_binary_uuid` argument. Accepted file layouts and read
semantics are unchanged.
### Validation
- Repository clang-format 16 check, build hygiene checks, and `git diff
--check` passed.
- Ported the three diagnostic unit tests, covering required/repeated
fallback leaves, incorrect physical types, group fallback values, a
valid control, unsupported Variant versions, and truncated footers.
- The source PR passed all 45 `ParquetSchemaTest.*` tests in a focused
ASAN binary on branch-4.1. The master port has not been compiled or
unit-tested locally; master build/test validation is pending CI.
### Release note
Parquet V2 footer errors now identify the input file. Invalid Variant
fallback errors additionally describe the field path and actual schema
declaration.
### Check List (For Author)
- Test
- [x] Unit Test (ported; master execution pending CI)
- Behavior changed:
- [x] Yes. Diagnostic context is added; status codes and validation
rules are preserved.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/format_v2/parquet/native_schema_desc.cpp | 40 +++++++-
be/src/format_v2/parquet/parquet_file_context.cpp | 10 +-
be/test/format_v2/parquet/parquet_schema_test.cpp | 118 ++++++++++++++++++++++
3 files changed, 164 insertions(+), 4 deletions(-)
diff --git a/be/src/format_v2/parquet/native_schema_desc.cpp
b/be/src/format_v2/parquet/native_schema_desc.cpp
index 19180c5c86c..5a9812d3841 100644
--- a/be/src/format_v2/parquet/native_schema_desc.cpp
+++ b/be/src/format_v2/parquet/native_schema_desc.cpp
@@ -422,9 +422,45 @@ Status validate_variant_layout(const NativeFieldSchema&
group_field,
if (fallback != nullptr &&
(!fallback->children.empty() || fallback->physical_type !=
tparquet::Type::BYTE_ARRAY ||
!valid_fallback_repetition)) {
+ // Resolve the Variant-relative path only on failure; valid
schemas need no extra
+ // per-node path allocation just to distinguish repeated wrapper
names such as element.
+ std::string path;
+ std::function<bool(const NativeFieldSchema&)> find_path =
+ [&](const NativeFieldSchema& field) {
+ const auto prefix_size = path.size();
+ if (!path.empty()) {
+ path += '.';
+ }
+ path += field.name;
+ if (&field == fallback) {
+ return true;
+ }
+ for (const auto& child : field.children) {
+ if (find_path(child)) {
+ return true;
+ }
+ }
+ path.resize(prefix_size);
+ return false;
+ };
+ find_path(group_field);
+ const auto* physical = &fallback->parquet_schema;
+ // A repeated primitive is normalized into an array whose child
retains the original
+ // SchemaElement. Report that declaration rather than the
synthetic array wrapper.
+ if (!physical->__isset.repetition_type &&
fallback->children.size() == 1 &&
+ fallback->children[0].parquet_schema.repetition_type ==
+ tparquet::FieldRepetitionType::REPEATED) {
+ physical = &fallback->children[0].parquet_schema;
+ }
return Status::Corruption(
- "Parquet Variant wrapper {} value must be an {}
BYTE_ARRAY", wrapper.name,
- allow_required_fallback ? "optional or required" :
"optional");
+ "Parquet Variant wrapper {} value must be an {}
BYTE_ARRAY; "
+ "variant_path={}, physical_type={}, repetition={},
children={}",
+ wrapper.name, allow_required_fallback ? "optional or
required" : "optional",
+ path, physical->__isset.type ?
tparquet::to_string(physical->type) : "GROUP",
+ physical->__isset.repetition_type
+ ? tparquet::to_string(physical->repetition_type)
+ : "UNSET",
+ physical->__isset.num_children ? physical->num_children :
0);
}
if (typed != nullptr) {
if (!typed->parquet_schema.__isset.repetition_type ||
diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp
b/be/src/format_v2/parquet/parquet_file_context.cpp
index 8d0f23e13b2..9510abb72d6 100644
--- a/be/src/format_v2/parquet/parquet_file_context.cpp
+++ b/be/src/format_v2/parquet/parquet_file_context.cpp
@@ -306,9 +306,15 @@ Status ParquetFileContext::open(io::FileReaderSPtr
input_file_reader, io::IOCont
native_metadata =
native_meta_cache_handle.data<NativeParquetMetadata>();
++native_footer_cache_hits;
} else {
- RETURN_IF_ERROR(parse_native_parquet_footer(
+ auto footer_status = parse_native_parquet_footer(
native_file, &native_metadata_owner, &native_footer_size,
io_ctx,
- enable_mapping_varbinary, enable_mapping_timestamp_tz,
preserve_binary_uuid));
+ enable_mapping_varbinary, enable_mapping_timestamp_tz,
preserve_binary_uuid);
+ if (!footer_status.ok()) {
+ // Schema validators have no file identity. Attach it before
leaving the file
+ // layer, preserving the original code used by callers to classify
the failure.
+ return footer_status.prepend(fmt::format("Failed to read Parquet
footer for '{}': ",
+
native_file->path().native()));
+ }
++native_footer_read_calls;
if (has_stable_meta_cache_identity && meta_cache != nullptr &&
meta_cache->enabled()) {
meta_cache->insert(meta_cache_key, native_metadata_owner.release(),
diff --git a/be/test/format_v2/parquet/parquet_schema_test.cpp
b/be/test/format_v2/parquet/parquet_schema_test.cpp
index f7f96ea2c7e..ceef0795743 100644
--- a/be/test/format_v2/parquet/parquet_schema_test.cpp
+++ b/be/test/format_v2/parquet/parquet_schema_test.cpp
@@ -17,6 +17,9 @@
#include <gtest/gtest.h>
+#include <algorithm>
+#include <array>
+#include <cstring>
#include <limits>
#include <optional>
#include <string>
@@ -37,6 +40,10 @@
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_file_context.h"
#include "format_v2/parquet/reader/native_column_reader.h"
+#include "io/fs/file_reader.h"
+#include "io/io_common.h"
+#include "util/coding.h"
+#include "util/thrift_util.h"
namespace doris::format::parquet {
@@ -221,6 +228,117 @@ std::vector<tparquet::SchemaElement>
shredded_array_variant_schema(bool include_
} // namespace
+namespace {
+
+class FooterDiagnosticFileReader final : public io::FileReader {
+public:
+ explicit FooterDiagnosticFileReader(std::string bytes) :
_bytes(std::move(bytes)) {}
+ Status close() override { return Status::OK(); }
+ const io::Path& path() const override { return _path; }
+ size_t size() const override { return _bytes.size(); }
+ bool closed() const override { return false; }
+ int64_t mtime() const override { return 0; }
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext*) override {
+ *bytes_read = offset < _bytes.size() ? std::min(result.size,
_bytes.size() - offset) : 0;
+ if (*bytes_read != 0) {
+ memcpy(result.data, _bytes.data() + offset, *bytes_read);
+ }
+ return Status::OK();
+ }
+
+private:
+ io::Path _path {"s3://example-bucket/nested.parquet"};
+ std::string _bytes;
+};
+
+Status open_diagnostic_footer(const std::vector<tparquet::SchemaElement>&
schema) {
+ tparquet::FileMetaData metadata;
+ metadata.__set_version(1);
+ metadata.__set_schema(schema);
+ metadata.__set_num_rows(0);
+ metadata.__set_row_groups({});
+ std::vector<uint8_t> footer;
+ ThriftSerializer serializer(true, 1024);
+ RETURN_IF_ERROR(serializer.serialize(&metadata, &footer));
+ std::string bytes = "PAR1";
+ bytes.append(reinterpret_cast<const char*>(footer.data()), footer.size());
+ std::array<uint8_t, 4> length;
+ encode_fixed32_le(length.data(), static_cast<uint32_t>(footer.size()));
+ bytes.append(reinterpret_cast<const char*>(length.data()), length.size());
+ bytes.append("PAR1");
+ auto file = std::make_shared<FooterDiagnosticFileReader>(std::move(bytes));
+ ParquetFileContext context;
+ return context.open(file, nullptr, false, io::FileDescription(), false,
false, false);
+}
+
+} // namespace
+
+TEST(ParquetSchemaTest, VariantFallbackErrorReportsPathAndActualSchema) {
+ auto schema = shredded_array_variant_schema(true, false);
+ schema[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ auto status = open_diagnostic_footer(schema);
+ ASSERT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ EXPECT_NE(status.to_string().find("s3://example-bucket/nested.parquet"),
std::string::npos);
+
EXPECT_NE(status.to_string().find("variant_path=payload.typed_value.element.value"),
+ std::string::npos);
+ EXPECT_NE(status.to_string().find("physical_type=BYTE_ARRAY"),
std::string::npos);
+ EXPECT_NE(status.to_string().find("repetition=REQUIRED"),
std::string::npos);
+ EXPECT_NE(status.to_string().find("children=0"), std::string::npos);
+
+ schema[7].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
+ ASSERT_TRUE(open_diagnostic_footer(schema).ok());
+ schema[7].__set_type(tparquet::Type::INT64);
+ status = open_diagnostic_footer(schema);
+ ASSERT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ EXPECT_NE(status.to_string().find("physical_type=INT64"),
std::string::npos);
+ EXPECT_NE(status.to_string().find("repetition=OPTIONAL"),
std::string::npos);
+
+ schema[7].__set_type(tparquet::Type::BYTE_ARRAY);
+ schema[7].__set_repetition_type(tparquet::FieldRepetitionType::REPEATED);
+ status = open_diagnostic_footer(schema);
+ ASSERT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+ EXPECT_NE(status.to_string().find("physical_type=BYTE_ARRAY"),
std::string::npos);
+ EXPECT_NE(status.to_string().find("repetition=REPEATED"),
std::string::npos);
+}
+
+TEST(ParquetSchemaTest, VariantFallbackGroupErrorReportsActualChildren) {
+ auto schema = shredded_object_variant_schema();
+ auto child = schema[6];
+ child.__set_name("child");
+ schema[6].__isset.type = false;
+ schema[6].__set_num_children(1);
+ schema.insert(schema.begin() + 7, child);
+ const auto status = open_diagnostic_footer(schema);
+ ASSERT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
+
EXPECT_NE(status.to_string().find("variant_path=payload.typed_value.field.value"),
+ std::string::npos);
+ EXPECT_NE(status.to_string().find("physical_type=GROUP"),
std::string::npos);
+ EXPECT_NE(status.to_string().find("children=1"), std::string::npos);
+}
+
+TEST(ParquetSchemaTest, FooterContextPreservesUnsupportedAndCorruptionCodes) {
+ auto schema = unshredded_variant_schema(2);
+ NativeFieldDescriptor descriptor;
+ const auto expected = descriptor.parse_from_thrift(schema);
+ ASSERT_FALSE(expected.ok());
+ const auto unsupported = open_diagnostic_footer(schema);
+ EXPECT_EQ(unsupported.code(), expected.code());
+ EXPECT_NE(unsupported.to_string().find(expected.msg()), std::string::npos);
+
EXPECT_NE(unsupported.to_string().find("s3://example-bucket/nested.parquet"),
+ std::string::npos);
+
+ auto file = std::make_shared<FooterDiagnosticFileReader>("PAR1");
+ ParquetFileContext context;
+ const auto corrupt =
+ context.open(file, nullptr, false, io::FileDescription(), false,
false, false);
+ ASSERT_TRUE(corrupt.is<ErrorCode::CORRUPTION>()) << corrupt;
+ EXPECT_NE(corrupt.to_string().find("s3://example-bucket/nested.parquet"),
std::string::npos);
+ EXPECT_NE(corrupt.to_string().find("too small for a footer"),
std::string::npos);
+}
+
TEST(ParquetSchemaTest, NativeSchemaRecognizesVariantLogicalGroup) {
for (const auto version : {std::optional<int8_t> {}, std::optional<int8_t>
{1}}) {
NativeFieldDescriptor descriptor;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
