[
https://issues.apache.org/jira/browse/IMPALA-15443?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zoltán Borók-Nagy resolved IMPALA-15443.
----------------------------------------
Resolution: Fixed
> 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
> Assignee: Zoltán Borók-Nagy
> Priority: Major
> Labels: correctness
> Fix For: Impala 5.0.0
>
>
> 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}}.
> h3. 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
(v10.3.26#10030026)
