Zoltán Borók-Nagy created IMPALA-15443:
------------------------------------------
Summary: Parquet stats filtering decodes INT32/INT64 min/max at
the table's wider DECIMAL width (decimal widening, int->decimal), silently
dropping rows
Key: IMPALA-15443
URL: https://issues.apache.org/jira/browse/IMPALA-15443
Project: IMPALA
Issue Type: Bug
Components: Backend
Reporter: Zoltán Borók-Nagy
h3. Problem
If a Parquet file stores a column as INT32 or INT64, but the table's DECIMAL
type needs more bytes (INT32 file read as precision >= 10, INT64 file read as
precision >= 19), Impala decodes the file's min/max statistics at the *table's*
width. The upper bytes of the decoded value are not taken from the file, so
they depend on memory contents and answers can differ between executions.
Row-group statistics and the page index (column index) both use these values,
so row groups and pages that contain matching rows are skipped and queries
silently return too few rows. With default options, some tables gave the same
wrong answer in every try and others varied. With negative values the answer
was wrong in almost every try (4.5.2 and master; one master run of {{t_neg}}
with {{PARQUET_READ_PAGE_INDEX=false}} was right in 2 of 50 tries).
This is the normal state after a decimal precision widening:
* Iceberg type promotion {{decimal(10,2) -> decimal(20,2)}} (spec-valid). Old
files keep INT64 decimal(10,2).
* Plain Parquet tables after {{ALTER TABLE ... CHANGE amt amt DECIMAL(20,2)}}
(the IMPALA-7087 path).
* Master (5.0.0) only: INT/BIGINT columns read as DECIMAL (IMPALA-13625), e.g.
after {{ALTER TABLE t CHANGE c c DECIMAL(10,0)}}. IMPALA-13625 requires
precision >= 10 for INT32 and >= 19 for INT64, so every such read uses a wider
slot.
Only files that store the column as INT32/INT64 *and* carry column orders are
affected: parquet-mr 1.17.1 (as used by Spark 4.1.3 + Iceberg 1.11.0), pyarrow
with {{store_decimal_as_integer=True}}, and on master Impala-written INT/BIGINT
columns read as DECIMAL through IMPALA-13625. Files without column orders are
not affected, because their decimal stats are ignored. Impala-written DECIMAL
columns (FIXED_LEN_BYTE_ARRAY) are not affected. The values themselves are read
correctly; only stats-based skipping is wrong.
h3. Steps to reproduce (no Iceberg needed)
{code:python}
import decimal, pyarrow as pa, pyarrow.parquet as pq
vals = [decimal.Decimal(f"{i}.00") for i in range(-5, 5)]
t = pa.table({"id": pa.array(range(1, 11), pa.int64()), "amt": pa.array(vals,
pa.decimal128(10, 2))})
pq.write_table(t, "decimal_int64_negative.parquet",
store_decimal_as_integer=True,
write_page_index=True, max_rows_per_page=2,
use_dictionary=False, compression="NONE")
# amt: INT64 Decimal(precision=10, scale=2), min -5.00, max 4.00
{code}
Put the file into an empty directory, then in Impala:
{code:sql}
CREATE EXTERNAL TABLE hive_neg (id BIGINT, amt DECIMAL(20,2)) STORED AS PARQUET
LOCATION '<dir>';
SELECT count(*) FROM hive_neg WHERE amt < 0; -- expected 5
SELECT count(*) FROM hive_neg WHERE amt = -3.00; -- expected 1
{code}
Iceberg variant (Spark SQL, then register the table in Impala):
{code:sql}
CREATE TABLE t_neg (id BIGINT, amt DECIMAL(10,2)) USING iceberg TBLPROPERTIES
('format-version'='2');
INSERT INTO t_neg SELECT id, CAST(id AS DECIMAL(10,2)) FROM range(-5, 5, 1, 1);
-- INT64 decimal(10,2)
ALTER TABLE t_neg ALTER COLUMN amt TYPE DECIMAL(20,2);
INSERT INTO t_neg SELECT id, CAST(id AS DECIMAL(20,2)) FROM range(-5, 5, 1, 1);
-- FIXED_LEN_BYTE_ARRAY decimal(20,2)
-- Spark: SELECT count(*) FROM t_neg WHERE amt < 0 -> 10
{code}
{code:sql}
-- Impala, for a Hadoop-catalog table (use the HMS equivalent otherwise)
CREATE EXTERNAL TABLE t_neg STORED AS ICEBERG LOCATION '<table dir>'
TBLPROPERTIES('iceberg.catalog'='hadoop.tables');
SELECT count(*) FROM t_neg WHERE amt < 0; -- expected 10
{code}
h3. Actual result (Impala 4.5.2; master 3d4d47dc30 RELEASE build gives the same
unless noted)
* {{hive_neg}}: {{amt < 0}} returns *0* (50/50 tries, also on master), {{amt =
-3.00}} returns *0*. {{SELECT *}} returns the correct values. With
{{PARQUET_READ_STATISTICS=false}} the answers are 5 and 1.
* {{t_neg}}: {{amt < 0}} returns *5* instead of 10 (50/50 on master); {{amt + 0
< 0}} (no stats conjunct) returns 10. Profile: {{NumRowGroups: 2}},
{{NumStatsFilteredRowGroups: 1}}. {{amt = -3.00}} returns 1 instead of 2.
* On a larger Iceberg table (100k rows per file, 5 pages per column chunk),
{{amt < 50000.00}} returns 50278 instead of 100260 (the whole old file is
skipped). On 4.5.2 the answer varied between executions of the same query: in
one try only 4 of the 5 pages of the old file were skipped (60274), and the
original benchmark query returned 4 different answers in 4 tries.
* Master: two independently generated tables of this shape skipped the whole
old file in every default-options try (100/100 and 330/330); with
{{PARQUET_READ_PAGE_INDEX=false}} they were wrong in 12/50 and 19/100 tries.
The 4-of-5-page skip showed up on another promoted column (INT32 decimal(9,2)
-> DECIMAL(12,2): 59719 instead of 99697, {{NumStatsFilteredPages: 4}}). A
5-file table shaped like the benchmark returned 4 different wrong answers in 40
tries, never the right one.
* {{PARQUET_READ_PAGE_INDEX=false}} does *not* fix it in general, because
row-group statistics are decoded the same way.
h3. Analysis (4.5.2 line numbers; master 3d4d47dc30 in brackets)
* {{ColumnStatsReader::ReadFromString()}}, case {{TYPE_DECIMAL}}
({{be/src/exec/parquet/parquet-column-stats.cc:164-177}} [167-180]), switches
on {{col_type_.GetByteSize()}}, i.e. the *table* type. For DECIMAL(20,2) it
calls {{DecodeDecimal<Decimal16Value>}} with the file's physical type INT64.
* That ends in the generic
{{ParquetPlainEncoder::DecodeNoBoundsCheck<Decimal16Value, INT64>}}
({{be/src/exec/parquet/parquet-common.h:468-478}} [469-479]), which does
{{memcpy(v, buffer, sizeof(InternalType))}}: it copies 16 bytes from an 8-byte
stat value. Its own comment says the generic template is only for no conversion
or narrowing conversions; there is no specialization for Decimal8Value/INT32,
Decimal16Value/INT32 or Decimal16Value/INT64.
* The extra bytes come from whatever follows the stat value in memory, most
likely the {{std::string}} small-string buffer (not confirmed in a debugger).
This explains the nondeterminism. For an 8-byte stat, byte 8 is then the
string's NUL terminator, so the upper word can never be the all-ones sign
extension, which is why negative values fail almost always.
* The data path is correct: {{CreateDecimalColumnReader()}}
({{parquet-column-readers.cc:1677}} [1895]) decodes by the physical type and
widens with {{ParquetDataConverter}}.
* {{ReadFromStringsBatch()}} (page-level min/max runtime filters) has the same
flaw, including for INT32 -> BIGINT and FLOAT -> DOUBLE
({{DecodeBatchOneBoundsCheckFastTrack<int64_t>/<double>}},
{{parquet-column-stats.cc:210/223}} [213/226]). It gives wrong join results
with explicit PAGE-level min/max filter settings, and also with *default* query
options when the widened column is the table's leading SORT BY column: with
{{MINMAX_FILTER_SORTED_COLUMNS=true}} (the default) Impala creates a min/max
filter for that column (IMPALA-10709), and the planner then runs it with
{{MINMAX_FILTER_THRESHOLD=0.5}} at PAGE level. Master, default options, plain
Parquet SORT BY table joined with {{VALUES (-3),(-2)}}: 0 rows instead of 2 in
30/30 tries for INT32 read as BIGINT, 30/30 for FLOAT read as DOUBLE, and 17/30
for DECIMAL(10,2) read as DECIMAL(20,2). Same-width controls were correct. An
Iceberg table promoted int -> long / float -> double with an Impala SORT BY
returned 2 instead of 4 in 20/20.
* Reachable since IMPALA-7087 / IMPALA-8131 (commit 9d46255739, 4.1.0). Before
that, {{ParquetMetadataUtils::ValidateColumn()}} rejected INT32/INT64 decimals
whose width differed from the table type; the table-width decode in
{{ReadFromString()}} is older. 4.1.0 through 4.4.x are likely affected but were
not tested. Master 3d4d47dc30 reproduces all steps above.
* IMPALA-13625 (reading integer columns as DECIMAL, 5.0.0) reaches the same
decode (reproduced on master). Impala-written table (id BIGINT, c1 INT, c2
BIGINT) with values -5..4, then {{ALTER TABLE ... CHANGE c1 c1 DECIMAL(10,0)}}
and {{CHANGE c2 c2 DECIMAL(19,0)}}: {{c1 < 0}} returns 0 (expected 5), {{c1 <
3}} returns 0 (expected 8), {{c2 < 0}} returns 0 (expected 5), 30/30 each.
Correct with {{PARQUET_READ_STATISTICS=false}}.
Suggested fix: decode decimal stats at the file's width (Decimal4Value for
INT32, Decimal8Value for INT64) and then {{ConvertSlot()}} to the table width,
like the column readers do; use the BIGINT/DOUBLE fast track only for matching
physical types; add {{DCHECK_LE(sizeof(InternalType), byte_size)}} to the
generic {{DecodeNoBoundsCheck}}.
h3. Workaround
{{SET PARQUET_READ_STATISTICS=false}} (also fixes the min/max runtime filter
join cases), or rewrite the old data files with the new type (verified with
Spark {{rewrite_data_files}}). For the sorted-column join case only,
{{MINMAX_FILTER_SORTED_COLUMNS=false}} also helps.
{{PARQUET_READ_PAGE_INDEX=false}} is not a workaround.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
