Zoltán Borók-Nagy created IMPALA-15374:
------------------------------------------
Summary: Late materialization for collections: SkipRows() does not
roll back current_row_, giving wrong FILE__POSITION and misaligned rows
Key: IMPALA-15374
URL: https://issues.apache.org/jira/browse/IMPALA-15374
Project: IMPALA
Issue Type: Bug
Components: Backend
Reporter: Zoltán Borók-Nagy
Assignee: Zoltán Borók-Nagy
{{BaseScalarColumnReader::SkipRows()}} undoes the one level that the
non-batched read protocol read ahead, but it never rolls back {{current_row_}}.
After every skip that follows a read, {{LastProcessedRow()}} is one too high,
and the error accumulates within the row group. Two user-visible symptoms
follow, and which one appears depends on whether the page index is used.
This affects collection columns read under late materialization, i.e. since
IMPALA-3841
(607bad042a, master only, not in a release). The same code is the blocker for
enabling late materialization for VARIANT (IMPALA-15142), which is how it was
found.
*Status: derived by tracing the code, not yet reproduced at runtime.* Repro
queries are below; they need a throwaway debug cluster.
h2. Root cause
{{BaseScalarColumnReader::NextLevels()}} (parquet-column-readers.cc:1461-1475)
advances
{{current_row_}} and sets {{levels_readahead_ = true}}:
{code:java}
if (rep_level_ == 0) ++current_row_; // max_rep_level() > 0
...
++current_row_; // max_rep_level() == 0
levels_readahead_ = true;
{code}
The undo in {{SkipRows()}} (parquet-column-readers.h:600-617) reverses
everything except that
increment:
{code:java}
if (levels_readahead_) {
rep_levels_.CachePrev();
def_levels_.CachePrev();
rep_level_ = ParquetLevel::INVALID_LEVEL;
def_level_ = ParquetLevel::INVALID_LEVEL;
++num_buffered_values_;
levels_readahead_ = false; // <-- current_row_ is never decremented
}
{code}
{{LastProcessedRow()}} is {{levels_readahead_ ? current_row_ - 1 :
current_row_}}
(parquet-column-readers.h:572-575), so clearing the flag alone shifts it by
one, and the
skip then counts from the shifted base. Nothing re-bases {{current_row_}} until
the next
row group ({{Reset()}}), or the next page under page filtering
({{StartPageFiltering()}},
parquet-column-readers.cc:1493).
h2. Symptom 1 (no page index): wrong FILE__POSITION
Values stay aligned, but the drift lands in FILE__POSITION, which
{{ReadFilePositionNonBatched()}} computes from {{LastProcessedRow()}}.
{code:sql}
set parquet_late_materialization_threshold=1;
select file__position, int_array, id from functional_parquet.complextypestbl
where id % 2 = 0;
-- correct (and what threshold=-1 returns): positions 1, 3, 5, 0
-- predicted with the bug: positions 1, 4, 7, 0 (7 exceeds
the 7-row file)
{code}
h2. Symptom 2 (page index active): collection values attached to the wrong row
In the page-filtering branch of {{SkipRowsInternal()}}
(parquet-column-readers.cc:1806-1812)
the number of rows to skip is {{skip_row_id - LastProcessedRow()}}, so the
reader skips one row
too few and ends up physically one row behind while its row *labels* stay in
sync. When exactly
one row must be skipped, the {{last_row < skip_row_id}} guard is false and
nothing is skipped
at all.
{code:sql}
-- table t loaded from
testdata/data/customer_nested_multiblock_multipage.parquet
-- (3 row groups x 100 rows; every column has 5 pages of 20 rows; page index
present)
set batch_size=4;
set expand_complex_types=true;
select c_custkey, c_orders from t where c_custkey > 280 and c_custkey % 9 = 2;
-- predicted with the bug: 281 correct, 290 shows customer 289's orders,
-- 299 shows customer 297's orders; no error is raised
-- compare against: set parquet_late_materialization_threshold=-1; (or
parquet_read_page_index=false)
{code}
Nothing catches this: {{CheckPageFiltering()}} compares row labels, which stay
in sync, and
tolerates a difference of 1 anyway; the end of the row group is also reached by
label.
h2. Why no existing test sees it
* {{nested-types-parquet-page-index.test}} uses inner joins ({{n, n.arr}}),
where the planner
adds an {{!empty()}} predicate that makes the collection a filter reader with
{{max_rep_level() > 0}}, so late materialization is off.
* The two collection + FILE__POSITION queries in
{{virtual-column-file-position-parquet.test}}
have the collection as the only reader, so {{non_filter_readers_}} is empty
and late
materialization is off.
* {{... where c_custkey = 300}} in the same file does combine collections, the
page index and
result checks, but with the default {{batch_size=1024}} the whole 20-row
candidate range fits
in one scratch batch, so {{SkipRows()}} is never called after a read and the
undo never runs.
* {{parquet-late-materialization*.test}} uses {{parquet_read_page_index=false}}
or non-pushable
predicates and asserts counters or a constant, not collection values.
h2. Suggested fix
In the undo, before the levels are invalidated:
{code:java}
// SkipRows() is only ever called on a top-level row boundary.
DCHECK(max_rep_level() == 0 || rep_level_ == 0);
--current_row_;
{code}
A {{DCHECK_EQ(LastProcessedRow(), last_processed_row_before_undo)}} at the end
of the undo
makes the invariant explicit and would have caught this.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
