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 1bb35373ee0 [fix](be) Prepare Parquet null maps before logical
conversion (#69448)
1bb35373ee0 is described below
commit 1bb35373ee076af701e9070d472f7ee47073b565
Author: Socrates <[email protected]>
AuthorDate: Fri Oct 23 21:27:29 2026 +0800
[fix](be) Prepare Parquet null maps before logical conversion (#69448)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #63001
Problem Summary:
A Parquet schema conversion can write past an empty destination null map
and crash the BE. For example, a TVF infers `BIGINT` from one file and
then reads `STRING` value `"xx"` from another file.
`CastStringConverter` marks the failed cast as NULL, but
`PhysicalToLogicalConverter` currently copies the source null map only
after that conversion returns.
Prepare the appended destination null-map slice before logical
conversion. This preserves source NULLs and the existing destination
prefix, lets converters mark additional failed casts, and retains COW
isolation. The same boundary also covers other logical conversions that
produce NULL, such as numeric overflow. Non-nullable conversion errors
remain unchanged.
Coverage added:
- Converter unit tests for BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY, nullable
and non-nullable sources, invalid and valid strings, source NULLs, empty
destinations, successive append batches, COW aliases, non-nullable
errors, and numeric overflow.
- Extend the existing `test_s3_tvf` suite with a mixed-schema Parquet
regression: generate a BIGINT schema seed plus 4,096 STRING rows
containing invalid values, numeric strings and source NULLs, then
materialize all rows with both File Scanner V1 and V2 using
`batch_size=1024` and non-strict casts. Compare their complete ordered
results, including NULL positions. This differential check reuses the
same S3 fixtures and needs no new golden output; fixed expected
conversion values and null maps remain covered by the converter unit
tests. Remove the separate crash-only Groovy suite.
### Release note
Fix a BE crash when Parquet schema conversion produces NULL for an
invalid value.
### Check List (For Author)
- Test:
- [x] Unit tests added; not executed locally.
- [x] V1/V2 end-to-end differential regression added to the existing S3
TVF suite; not executed locally.
- Passed: build hygiene, clang-format 16 formatting/check, `git diff
--check`, Groovy syntax compilation.
- `bash run-be-ut.sh --run --filter='ParquetColumnConvertTest.*' -j 4`
was stopped while downloading the apache-orc submodule. The local
third-party installation also lacks `libhs.a`; no test executable was
produced.
- `./run-regression-test.sh --run -d external_table_p0/tvf -s
test_s3_tvf` failed before executing any suite: Maven 3.10 rejects
existing duplicate Gson declarations in the regression framework POM.
- `run-clang-tidy.sh` could not run because no compilation database was
generated. Runtime and sanitizer validation remain for CI.
- Behavior changed:
- [x] Yes. Invalid nullable conversions produce NULL without accessing
an unallocated null map.
- Does this need documentation?
- [x] No.
<details>
<summary>Self-review checkpoints</summary>
- Goal: prepares the null map before any logical-converter failure
write; the new unit cases reproduce that boundary. Runtime validation is
pending.
- Scope: one conversion boundary changed, with focused unit and crash
regression tests.
- Concurrency and locks: no new shared state or locking; columns belong
to the reader, and destination aliases are detached using existing COW
helpers.
- Lifecycle and static initialization: no new lifetime mechanism or
static state; source null maps remain independently owned.
- Configuration: no new configuration; the regression restores all three
changed session settings (`enable_file_scanner_v2`, `batch_size`,
`enable_strict_cast`).
- Compatibility: no symbol, protocol or storage-format changes.
- Parallel paths: ORC already aligns destination nulls before logical
conversion; File Scanner V2 uses a separate CAST implementation with
allocated null maps and is now covered by the same end-to-end
differential test. Consistent Parquet conversion paths are unchanged.
- Conditions: no new production conditional branch; existing nullable
handling is reused.
- Coverage: source and conversion NULLs, valid values, two physical
representations, successive batches, COW and non-nullable errors are
covered; runtime execution is explicitly unverified.
- Expected results: checked directly in converter unit tests; the
end-to-end test checks successful materialization and compares complete
V1/V2 results, including values and NULL positions, over multiple
batches.
- Observability: existing non-nullable conversion errors remain
sufficient; no additional per-row logging is needed.
- Transactions/persistence: not involved in the production change.
- Data writes/atomicity: no production write-path changes; the
regression writes only its unique temporary S3 prefix.
- FE/BE variables: none added.
- Performance: the existing null-map copy is moved before conversion,
retaining linear cost and avoiding a second merge pass.
- Other issues: no additional source-level issue found in the modified
path; CI compilation, UT/sanitizer and external regression execution
remain required.
</details>
### 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/parquet_column_convert.h | 10 +-
.../format/parquet/parquet_column_convert_test.cpp | 114 +++++++++++++++++++++
.../external_table_p0/tvf/test_s3_tvf.groovy | 42 ++++++++
3 files changed, 162 insertions(+), 4 deletions(-)
diff --git a/be/src/format/parquet/parquet_column_convert.h
b/be/src/format/parquet/parquet_column_convert.h
index 083711a5ce3..3374a3e2dc9 100644
--- a/be/src/format/parquet/parquet_column_convert.h
+++ b/be/src/format/parquet/parquet_column_convert.h
@@ -385,14 +385,16 @@ public:
get_appended_null_map_start(src_physical_col,
src_new_rows));
dst_logical_col = IColumn::mutate(std::move(dst_logical_col));
- const size_t dst_old_rows =
get_mutable_inner_column_size(dst_logical_col);
const size_t dst_old_null_map_size =
get_null_map_size_or_inner_column_size(dst_logical_col);
+ const size_t rows_to_convert =
get_mutable_inner_column_size(src_logical_column);
+ // Logical converters append all source rows and may mark failed casts
as NULL (e.g.
+ // "xx" -> BIGINT). Copy source nulls first so those writes have
allocated entries and
+ // are not overwritten after conversion. Keep any existing destination
prefix intact.
+ align_null_map(src_logical_column, dst_logical_col,
dst_old_null_map_size, rows_to_convert,
+ get_appended_null_map_start(src_logical_column,
rows_to_convert));
auto converted_column = dst_logical_col->assert_mutable();
RETURN_IF_ERROR(_logical_converter->convert(src_logical_column,
converted_column));
- const size_t dst_new_rows =
get_mutable_inner_column_size(dst_logical_col) - dst_old_rows;
- align_null_map(src_logical_column, dst_logical_col,
dst_old_null_map_size, dst_new_rows,
- get_appended_null_map_start(src_logical_column,
dst_new_rows));
return Status::OK();
}
diff --git a/be/test/format/parquet/parquet_column_convert_test.cpp
b/be/test/format/parquet/parquet_column_convert_test.cpp
index 0adc906ea9f..4b1db1c208a 100644
--- a/be/test/format/parquet/parquet_column_convert_test.cpp
+++ b/be/test/format/parquet/parquet_column_convert_test.cpp
@@ -424,6 +424,120 @@ TEST(ParquetColumnConvertTest,
EXPECT_EQ(original_dst->get_data_at(0).to_string(), "zz");
}
+// Exercise both a direct BYTE_ARRAY source and a physical-to-logical
temporary for
+// FIXED_LEN_BYTE_ARRAY. Both must prepare destination nulls before string
casts.
+TEST(ParquetColumnConvertTest, StringCastFailuresPreserveNullsAcrossBatches) {
+ for (const auto physical_type :
+ {tparquet::Type::BYTE_ARRAY, tparquet::Type::FIXED_LEN_BYTE_ARRAY}) {
+ for (const bool nullable_source : {false, true}) {
+ SCOPED_TRACE(testing::Message() << "physical_type=" <<
physical_type
+ << ", nullable_source=" <<
nullable_source);
+ FieldSchema field;
+ field.parquet_schema.__set_type(physical_type);
+ field.parquet_schema.__set_type_length(2);
+ field.data_type =
+ DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable_source);
+ const auto dst_type =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, true);
+ auto converter = PhysicalToLogicalConverter::get_converter(&field,
field.data_type,
+
dst_type, nullptr);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+ ColumnPtr dst = dst_type->create_column();
+ const std::vector<std::string> values {"xx", "05", "07", "09"};
+ const std::vector<uint8_t> expected_nulls {1, 0, 0, 0};
+ for (size_t batch = 0; batch < 2; ++batch) {
+ // Retain an alias to ensure preparing the null map respects
COW and does not
+ // modify an existing batch through another owner.
+ ColumnPtr dst_alias = dst;
+ MutableColumnPtr nested;
+ if (physical_type == tparquet::Type::BYTE_ARRAY) {
+ nested = ColumnString::create();
+ } else {
+ nested = ColumnFixedLengthObject::create(2);
+ }
+ auto nulls = ColumnUInt8::create();
+ for (size_t i = 0; i < values.size(); ++i) {
+ nested->insert_data(values[i].data(), values[i].size());
+ nulls->insert_value(i == 2);
+ }
+ ColumnPtr src;
+ if (nullable_source) {
+ src = ColumnNullable::create(std::move(nested),
std::move(nulls));
+ } else {
+ src = std::move(nested);
+ }
+ const ColumnPtr src_alias = src;
+ ASSERT_TRUE(converter->convert(src, field.data_type, dst_type,
dst, false).ok());
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*dst);
+ const auto& numbers = assert_cast<const
ColumnInt64&>(nullable.get_nested_column());
+ ASSERT_EQ(nullable.get_null_map_data().size(), (batch + 1) *
values.size());
+ ASSERT_EQ(numbers.size(), nullable.get_null_map_data().size());
+ for (size_t i = 0; i < numbers.size(); ++i) {
+ const size_t value_index = i % values.size();
+ EXPECT_EQ(nullable.get_null_map_data()[i],
+ expected_nulls[value_index] || (nullable_source
&& value_index == 2));
+ if (!nullable.get_null_map_data()[i]) {
+ EXPECT_EQ(numbers.get_data()[i],
std::stoi(values[value_index]));
+ }
+ }
+ EXPECT_EQ(dst_alias->size(), batch * values.size());
+ const auto& original = assert_cast<const
ColumnNullable&>(*dst_alias);
+ EXPECT_EQ(original.get_nested_column().size(), batch *
values.size());
+ if (nullable_source) {
+ const auto& source = assert_cast<const
ColumnNullable&>(*src_alias);
+ EXPECT_EQ(source.get_null_map_data()[0], 0);
+ EXPECT_EQ(source.get_null_map_data()[2], 1);
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetColumnConvertTest, InvalidStringToNonNullableBigIntReturnsError) {
+ FieldSchema field;
+ field.parquet_schema.__set_type(tparquet::Type::BYTE_ARRAY);
+ field.data_type =
DataTypeFactory::instance().create_data_type(TYPE_STRING, false);
+ const auto dst_type =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, false);
+ auto converter =
+ PhysicalToLogicalConverter::get_converter(&field, field.data_type,
dst_type, nullptr);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+ auto strings = ColumnString::create();
+ strings->insert_data("xx", 2);
+ ColumnPtr src = std::move(strings);
+ ColumnPtr dst = dst_type->create_column();
+ auto status = converter->convert(src, field.data_type, dst_type, dst,
false);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Failed to cast"), std::string::npos);
+}
+
+TEST(ParquetColumnConvertTest, NumericOverflowMarksNullableDestination) {
+ FieldSchema field;
+ field.parquet_schema.__set_type(tparquet::Type::INT64);
+ field.data_type =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, true);
+ const auto dst_type =
DataTypeFactory::instance().create_data_type(TYPE_FLOAT, true);
+ auto converter =
+ PhysicalToLogicalConverter::get_converter(&field, field.data_type,
dst_type, nullptr);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+ auto values = ColumnInt64::create();
+ values->insert_value(1LL << 40);
+ values->insert_value(7);
+ values->insert_value(9);
+ auto nulls = ColumnUInt8::create();
+ nulls->insert_value(0);
+ nulls->insert_value(0);
+ nulls->insert_value(1);
+ ColumnPtr src = ColumnNullable::create(std::move(values),
std::move(nulls));
+ ColumnPtr dst = dst_type->create_column();
+ ASSERT_TRUE(converter->convert(src, field.data_type, dst_type, dst,
false).ok());
+ const auto& nullable = assert_cast<const ColumnNullable&>(*dst);
+ ASSERT_EQ(nullable.get_null_map_data().size(), 3);
+ ASSERT_EQ(nullable.get_nested_column().size(), 3);
+ EXPECT_EQ(nullable.get_null_map_data()[0], 1);
+ EXPECT_EQ(nullable.get_null_map_data()[1], 0);
+ EXPECT_EQ(nullable.get_null_map_data()[2], 1);
+ EXPECT_FLOAT_EQ(assert_cast<const
ColumnFloat32&>(nullable.get_nested_column()).get_data()[1],
+ 7.0F);
+}
+
TEST(ParquetColumnConvertTest, RejectInvalidUuidPhysicalWidth) {
FieldSchema field;
field.parquet_schema.__set_type(tparquet::Type::FIXED_LEN_BYTE_ARRAY);
diff --git a/regression-test/suites/external_table_p0/tvf/test_s3_tvf.groovy
b/regression-test/suites/external_table_p0/tvf/test_s3_tvf.groovy
index 8ab9793c5d3..5848c396705 100644
--- a/regression-test/suites/external_table_p0/tvf/test_s3_tvf.groovy
+++ b/regression-test/suites/external_table_p0/tvf/test_s3_tvf.groovy
@@ -169,4 +169,46 @@ suite("test_s3_tvf", "p0,external") {
} finally {
}
+ // Exercise schema conversion with both readers, including failed casts,
source NULLs and
+ // multiple batches. The BIGINT seed must be listed before the STRING file
for TVF inference.
+ String castPrefix =
"s3://${getS3BucketName()}/regression/parquet_invalid_string_cast/${UUID.randomUUID()}"
+ String castStorageProperties = """
+ "s3.endpoint" = "${s3_endpoint}",
+ "s3.region" = "${region}",
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}"
+ """
+ sql """SELECT CAST(1 AS BIGINT) AS id, CAST('seed' AS STRING) AS name
+ INTO OUTFILE "${castPrefix}/00_seed/part_" FORMAT AS parquet
+ PROPERTIES (${castStorageProperties})"""
+ sql """SELECT CAST(CASE number % 4
+ WHEN 0 THEN 'xx'
+ WHEN 1 THEN '5'
+ WHEN 2 THEN NULL
+ ELSE '9' END AS STRING) AS id,
+ CAST(number AS STRING) AS name
+ FROM numbers("number" = "4096")
+ INTO OUTFILE "${castPrefix}/01_strings/part_" FORMAT AS parquet
+ PROPERTIES (${castStorageProperties})"""
+
+ def previousCastSettings = sql "SELECT @@enable_file_scanner_v2,
@@batch_size, @@enable_strict_cast"
+ try {
+ sql "SET batch_size=1024"
+ sql "SET enable_strict_cast=false"
+ // Materialize every value; COUNT(*) could use footer statistics and
skip conversion.
+ // Compare the actual reader outputs rather than hardcoding expected
rows in Groovy.
+ def castResults = [false, true].collect { scannerV2 ->
+ sql "SET enable_file_scanner_v2=${scannerV2}"
+ sql """SELECT id, name FROM S3(
+ "uri" = "${castPrefix}/{00_seed,01_strings}/*.parquet",
+ "format" = "parquet",
+ ${castStorageProperties}) ORDER BY name"""
+ }
+ assertEquals(castResults[0], castResults[1],
+ "V1 and V2 must preserve the same values and NULLs across
mixed-schema batches")
+ } finally {
+ sql "SET enable_file_scanner_v2=${previousCastSettings[0][0]}"
+ sql "SET batch_size=${previousCastSettings[0][1]}"
+ sql "SET enable_strict_cast=${previousCastSettings[0][2]}"
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
