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 a4504d3138f branch-4.1: [fix](parquet) Keep Variant metadata pruning
behind the safe conjunct prefix (#68144)
a4504d3138f is described below
commit a4504d3138f68e89f94c8111fa9ce41711bfed9a
Author: daidai <[email protected]>
AuthorDate: Fri Sep 18 13:59:15 2026 +0800
branch-4.1: [fix](parquet) Keep Variant metadata pruning behind the safe
conjunct prefix (#68144)
### What problem does this PR solve?
Issue Number: None
Related PR: #68137
Problem Summary:
On branch-4.1, a Variant predicate that follows an unsafe conjunct can
prune Parquet row groups and pages before that conjunct is evaluated, so
its error is lost:
```sql
SELECT COUNT(*) FROM iceberg_tbl
WHERE assert_true(id != 1, 'barrier') AND v['n'] > 5000;
-- returns 0 instead of failing with 'barrier'
```
Ordinary metadata pruning only uses the conjuncts before the first
unsafe one (`metadata_pruning_safe_conjunct_count`). The shredded
Variant row-group statistics, page-index pruning and
`has_variant_shredded_filter` still read every conjunct. This PR
restricts them to the same prefix, matching master.
As on master, CAST is not safe to pre-execute, so Variant typed-leaf
predicates no longer prune row groups or pages.
`test_iceberg_variant_read` takes the test updates from #68137 and adds
the error barrier case from master.
The partial-Variant write check also refreshes the table in Spark before
reading it back. Spark caches Iceberg snapshots, so it could still
report the pre-insert row count.
### Release note
Fix Iceberg Variant queries returning a result instead of an error when
a Variant predicate follows an error-raising expression such as
`assert_true`.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [ ] 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:
- [ ] No.
- [x] Yes. Variant typed-leaf predicates no longer prune Parquet row
groups or pages, as on master.
- 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_v2/parquet/parquet_statistics.cpp | 15 ++++-
.../format_v2/parquet/parquet_statistics_test.cpp | 21 ++++++
.../iceberg/test_iceberg_variant_read.groovy | 76 +++++++++++++---------
3 files changed, 79 insertions(+), 33 deletions(-)
diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp
b/be/src/format_v2/parquet/parquet_statistics.cpp
index 33db2a7b558..002288c402c 100644
--- a/be/src/format_v2/parquet/parquet_statistics.cpp
+++ b/be/src/format_v2/parquet/parquet_statistics.cpp
@@ -640,8 +640,15 @@ std::optional<VariantShreddedPredicate>
extract_variant_shredded_predicate(
.op = *op};
}
+VExprContextSPtrs metadata_pruning_conjuncts(const format::FileScanRequest&
request) {
+ const size_t safe_count =
+ std::min(request.metadata_pruning_safe_conjunct_count,
request.conjuncts.size());
+ return VExprContextSPtrs(request.conjuncts.begin(),
request.conjuncts.begin() + safe_count);
+}
+
bool has_variant_shredded_filter(const format::FileScanRequest& request) {
- return std::ranges::any_of(request.conjuncts, [](const auto& conjunct) {
+ const auto conjuncts = metadata_pruning_conjuncts(request);
+ return std::ranges::any_of(conjuncts, [](const auto& conjunct) {
return extract_variant_shredded_predicate(conjunct).has_value();
});
}
@@ -1078,7 +1085,9 @@ bool check_shredded_variant_statistics(
const tparquet::FileMetaData& metadata, const tparquet::RowGroup&
row_group,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
const format::FileScanRequest& request, const cctz::time_zone*
timezone) {
- for (const auto& conjunct : request.conjuncts) {
+ // A Variant predicate localized after an unsafe conjunct must not skip
the row group before
+ // that earlier conjunct reaches its row-level evaluation.
+ for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
const auto predicate = extract_variant_shredded_predicate(conjunct);
if (!predicate.has_value()) {
continue;
@@ -2060,7 +2069,7 @@ Status select_row_group_ranges_by_native_page_index(
}
}
- for (const auto& conjunct : request.conjuncts) {
+ for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
const auto predicate = extract_variant_shredded_predicate(conjunct);
if (!predicate.has_value()) {
continue;
diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp
b/be/test/format_v2/parquet/parquet_statistics_test.cpp
index d999c11fd0d..6f51499bc33 100644
--- a/be/test/format_v2/parquet/parquet_statistics_test.cpp
+++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp
@@ -1942,6 +1942,17 @@ TEST(NativeParquetStatisticsTest,
ShreddedVariantTypedValueDrivesPageFiltering)
.ok());
EXPECT_TRUE(selected_row_groups.empty());
+ // The same predicate can be localized after an earlier unsafe conjunct.
Metadata pruning must
+ // preserve that earlier expression's row-level error instead of skipping
the whole row group.
+ request.metadata_pruning_safe_conjunct_count = 0;
+ ASSERT_TRUE(format::parquet::select_row_groups_by_metadata(
+ footer_only_metadata, schema, request, nullptr,
&selected_row_groups, false,
+ nullptr, nullptr, nullptr, nullptr, {},
+ format::parquet::ParquetMetadataProbeMode::FOOTER_ONLY)
+ .ok());
+ EXPECT_EQ(selected_row_groups, std::vector<int>({0}));
+ request.metadata_pruning_safe_conjunct_count =
std::numeric_limits<size_t>::max();
+
auto leaf_projection = format::LocalColumnIndex::partial_local(0);
auto typed_object_projection = format::LocalColumnIndex::partial_local(2);
auto field_projection = format::LocalColumnIndex::partial_local(0);
@@ -2054,6 +2065,16 @@ TEST(NativeParquetStatisticsTest,
ShreddedVariantTypedValueDrivesPageFiltering)
EXPECT_EQ(pruning_stats.page_index_read_calls, 1);
EXPECT_EQ(pruning_stats.filtered_page_rows, 50);
+ request.metadata_pruning_safe_conjunct_count = 0;
+ ASSERT_TRUE(format::parquet::select_row_group_ranges_by_native_page_index(
+ metadata, metadata.row_groups[0], page_indexes,
schema, request, 100,
+ &selected_ranges, &skip_plans, nullptr)
+ .ok());
+ ASSERT_EQ(selected_ranges.size(), 1);
+ EXPECT_EQ(selected_ranges[0].start, 0);
+ EXPECT_EQ(selected_ranges[0].length, 100);
+ request.metadata_pruning_safe_conjunct_count =
std::numeric_limits<size_t>::max();
+
auto row_group_with_root_residual = metadata.row_groups[0];
row_group_with_root_residual.columns[1].meta_data.statistics.__set_null_count(99);
ASSERT_TRUE(format::parquet::select_row_group_ranges_by_native_page_index(
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
index eb723df128f..bbc0d3977a1 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
@@ -1225,12 +1225,20 @@ public class AppendVariantEqualityDelete {
if (positiveCounters.every { String counter -> counterSum(lastProfile,
counter) > 0 }) {
return lastProfile
}
- return profileAction.waitProfile({
- lastProfile = profileAction.getProfileBySql(token,
positiveCounters)
- return positiveCounters.every {
- String counter -> counterSum(lastProfile, counter) > 0
- } ? lastProfile : ""
- }, [], "Completed profile with positive counters ${positiveCounters}
for ${token}")
+ try {
+ return profileAction.waitProfile({
+ lastProfile = profileAction.getProfileBySql(token,
positiveCounters)
+ return positiveCounters.every {
+ String counter -> counterSum(lastProfile, counter) > 0
+ } ? lastProfile : ""
+ }, [], "Completed profile with positive counters
${positiveCounters} for ${token}")
+ } catch (IllegalStateException e) {
+ // The wait only reports an empty profile, so name the counters
that stayed at zero.
+ Map<String, Long> sums = positiveCounters.collectEntries { String
counter ->
+ [(counter): counterSum(lastProfile, counter)]
+ }
+ throw new IllegalStateException("${e.getMessage()}counter sums:
${sums}", e)
+ }
}
String evolutionInitial = latestSnapshotId("variant_evolution")
@@ -1398,8 +1406,8 @@ public class AppendVariantEqualityDelete {
WHERE v['shared'] >= 20
ORDER BY id
"""
- // The stable snapshot contributes a genuinely shredded file, while the
appended file uses
- // the unshredded fallback. More than four rows qualify, forcing local
TopN overshoot to be
+ // The stable snapshot contributes a genuinely shredded file, while the
appended file is read by
+ // seeking its unshredded value. More than four rows qualify, forcing
local TopN overshoot to be
// truncated after the merge exchange while the mapper-eligible projected
path crosses the wire.
explain {
sql """
@@ -1431,11 +1439,11 @@ public class AppendVariantEqualityDelete {
"""
assertEquals(4, projectedGatherRows.size())
String projectedGatherProfile = getProfileByToken(projectedGatherToken,
- ["VariantLeafProjections",
"VariantDirectLeafPathMisses"]).toString()
+ ["VariantLeafProjections",
"VariantUnshreddedDirectSeekRows"]).toString()
assertTrue(counterSum(projectedGatherProfile, "VariantLeafProjections") >
0,
"The projected TopN did not read a physical shredded Variant leaf")
- assertTrue(counterSum(projectedGatherProfile,
"VariantDirectLeafPathMisses") > 0,
- "The projected TopN did not combine the unshredded fallback file")
+ assertTrue(counterSum(projectedGatherProfile,
"VariantUnshreddedDirectSeekRows") > 0,
+ "The projected TopN did not combine the unshredded Variant file")
order_qt_variant_projected_remote_gather """
SELECT id,
CAST(projected['n'] AS INT)
@@ -1489,14 +1497,12 @@ public class AppendVariantEqualityDelete {
WHERE CAST(v['n'] AS INT) >= 8000
"""
String multiRowGroupColdProfile = getProfileByToken(multiRowGroupColdToken,
- ["RowGroupsTotalNum", "VariantDirectLeafPathMisses",
"VariantReconstructedRows",
+ ["RowGroupsTotalNum", "VariantUnshreddedDirectSeekRows",
"FilteredRowsByLazyRead"]).toString()
assertTrue(counterSum(multiRowGroupColdProfile, "RowGroupsTotalNum") > 1,
"The generated Variant file did not contain multiple Parquet
row groups")
- assertTrue(counterSum(multiRowGroupColdProfile,
"VariantDirectLeafPathMisses") > 0,
- "The unshredded scan did not record its direct-leaf fallback")
- assertTrue(counterSum(multiRowGroupColdProfile,
"VariantReconstructedRows") > 0,
- "The unshredded scan did not reconstruct Variant rows")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantUnshreddedDirectSeekRows") > 0,
+ "The unshredded scan did not seek its predicate leaf")
assertTrue(counterSum(multiRowGroupColdProfile, "FilteredRowsByLazyRead")
> 0,
"The unshredded Variant predicate did not defer non-predicate
columns")
String multiRowGroupWarmToken =
@@ -1507,9 +1513,9 @@ public class AppendVariantEqualityDelete {
WHERE CAST(v['n'] AS INT) >= 8000
"""
String multiRowGroupWarmProfile = getProfileByToken(multiRowGroupWarmToken,
- ["VariantDirectLeafPathMisses"]).toString()
- assertTrue(counterSum(multiRowGroupWarmProfile,
"VariantDirectLeafPathMisses") > 0,
- "The warm unshredded scan did not preserve its direct-leaf
fallback")
+ ["VariantUnshreddedDirectSeekRows"]).toString()
+ assertTrue(counterSum(multiRowGroupWarmProfile,
"VariantUnshreddedDirectSeekRows") > 0,
+ "The warm unshredded scan did not seek its predicate leaf")
qt_variant_multi_row_group_result """
SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
FROM variant_multi_row_group
@@ -1583,7 +1589,8 @@ public class AppendVariantEqualityDelete {
"The shredded predicate did not defer complete Variant output")
// The query projects the complete Variant while its predicate reads the
shredded typed leaf.
- // The appended unshredded file must fall back independently in the same
scan.
+ // The appended unshredded file must be read independently in the same
scan. CAST is not safe
+ // to pre-execute, so the metadata-pruning fence keeps this predicate out
of page pruning.
String pagePruningToken = "iceberg_variant_page_pruning_" +
UUID.randomUUID().toString()
sql """
SELECT '${pagePruningToken}', id, CAST(v AS STRING)
@@ -1592,16 +1599,14 @@ public class AppendVariantEqualityDelete {
ORDER BY id
"""
String pagePruningProfile = getProfileByToken(pagePruningToken,
- ["FilteredRowsByPage", "VariantLeafProjections",
"VariantDirectLeafPathMisses",
+ ["VariantLeafProjections", "VariantUnshreddedDirectSeekRows",
"VariantDirectLeafRows", "VariantReconstructedRows"]).toString()
- assertTrue(counterSum(pagePruningProfile, "FilteredRowsByPage") > 0,
- "Shredded Variant typed_value did not filter any Parquet page")
// The predicate_access_paths contract keeps the typed leaf eager while
the complete Variant
// root is read through the independent deferred-output projection.
assertTrue(counterSum(pagePruningProfile, "VariantLeafProjections") > 0,
"A root Variant output query did not retain its typed predicate
leaf projection")
- assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafPathMisses") >
0,
- "The mixed scan did not fall back for its unshredded Variant
file")
+ assertTrue(counterSum(pagePruningProfile,
"VariantUnshreddedDirectSeekRows") > 0,
+ "The mixed scan did not read its unshredded Variant file")
assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafRows") > 0,
"The mixed scan did not evaluate rows from the shredded typed
leaf")
assertTrue(counterSum(pagePruningProfile, "VariantReconstructedRows") > 0,
@@ -1623,6 +1628,17 @@ public class AppendVariantEqualityDelete {
WHERE CAST(v['n'] AS INT) > 3000
"""
+ // A later Variant metadata predicate must not prune away an earlier
error-producing conjunct.
+ test {
+ sql """
+ SELECT COUNT(*)
+ FROM variant_page_pruning
+ WHERE assert_true(id != 1, 'variant_metadata_error_barrier')
+ AND v['n'] > 5000
+ """
+ exception "variant_metadata_error_barrier"
+ }
+
order_qt_variant_aggregate """
SELECT CAST(v['ok'] AS BOOLEAN),
COUNT(*),
@@ -1761,6 +1777,8 @@ public class AppendVariantEqualityDelete {
SELECT id
FROM variant_write_guard FOR VERSION AS OF ${writeGuardSourceSnapshot}
"""
+ // Spark caches Iceberg snapshots, so refresh after Doris commits before
cross-engine reads.
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_write_guard"""
List<List<Object>> sparkPartialVariantRows = spark_iceberg """
SELECT COUNT(*), COUNT(payload)
FROM demo.${dbName}.variant_write_guard
@@ -1797,11 +1815,9 @@ public class AppendVariantEqualityDelete {
WHERE v['n'] >= 40
"""
String positionDeleteProfile = getProfileByToken(positionDeleteToken,
- ["VariantDirectLeafPathMisses",
"VariantReconstructedRows"]).toString()
- assertTrue(counterSum(positionDeleteProfile,
"VariantDirectLeafPathMisses") > 0,
- "Position-delete filtering did not preserve the unshredded
Variant fallback")
- assertTrue(counterSum(positionDeleteProfile, "VariantReconstructedRows") >
0,
- "Position-delete filtering did not reconstruct its Variant
rows")
+ ["VariantUnshreddedDirectSeekRows"]).toString()
+ assertTrue(counterSum(positionDeleteProfile,
"VariantUnshreddedDirectSeekRows") > 0,
+ "Position-delete filtering did not seek the unshredded Variant
leaf")
// A STRING path must build a physical leaf projection rather than falling
back to rebuilding
// the complete Variant. Reading only the accessed leaves is the entire
advantage shredded
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
