[
https://issues.apache.org/jira/browse/IMPALA-15374?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18118556#comment-18118556
]
ASF subversion and git services commented on IMPALA-15374:
----------------------------------------------------------
Commit 820e108c83ebc653a9b21fb54039586768f3b74e in impala's branch
refs/heads/master from Zoltan Borok-Nagy
[ https://gitbox.apache.org/repos/asf?p=impala.git;h=820e108c8 ]
IMPALA-15374: Roll back current_row_ when undoing a level read ahead
Collection column readers drive their children through the
non-batched interface, so a child always has one level read ahead.
When the scanner skips rows for such a child,
BaseScalarColumnReader::SkipRows() undoes that read ahead but did not
roll back 'current_row_', which NextLevels() had already incremented
for the level being un-read. SkipTopLevelRows() then counted from a
starting point one too high, so the counter drifted by one per
skipped range within a row group.
Which symptom this produces depends on the page index. Without it the
values stay aligned but LastProcessedRow() is too high, so a
collection reader that fills the file position slot reports positions
that are too large:
set parquet_late_materialization_threshold=1;
select file__position, int_array, id from complextypestbl
where id % 2 = 0;
returned positions 1, 4, 7, 0 instead of 1, 3, 5, 0; 7 is past the
end of the seven row file. With the page index the rows to skip are
computed as 'skip_row_id - LastProcessedRow()', so the reader skips
one row too few and falls behind the columns it is read with,
silently pairing rows with the wrong collection value:
set batch_size=4;
select c_custkey, count(o.o_orderkey), min(o.o_orderkey)
from customer_nested_multiblock_multipage c left join c.c_orders o
where c_custkey > 280 and c_custkey % 9 = 2 group by c_custkey;
returned customer 289's orders for customer 290, and none for
customer 299.
Roll the counter back under the same condition NextLevels() uses to
advance it, before 'rep_level_' is invalidated. Levels read ahead at
the end of a row group never advanced it and have 'rep_level_' ==
ROW_GROUP_END, so they are left alone.
Broken since IMPALA-3841, which added the read ahead undo along with
late materialization for collections. Only readers driven through the
non-batched interface are affected: top level scalar readers never
read a level ahead, and struct and VARIANT readers are still
excluded.
Testing:
- Added QueryTest/parquet-late-materialization-collections.test with
a query per symptom, run with late materialization and the page
index on and off. Both fail without the fix.
- Ran test_parquet_late_materialization.py, test_nested_types.py and
test_parquet_stats.py.
- customer_nested_multiblock_multipage is now loaded during dataload.
Change-Id: Icaf0aab770d87505513b0168f52f96480eda3202
Assisted-by: Claude Fable 5.1 (Claude Code)
Reviewed-on: http://gerrit.cloudera.org:8080/24886
Reviewed-by: Csaba Ringhofer <[email protected]>
Tested-by: Impala Public Jenkins <[email protected]>
> 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
> Priority: Major
>
> {{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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
