Zoltán Borók-Nagy created IMPALA-15444:
------------------------------------------
Summary: Iceberg: Parquet and ORC data files without field ids are
resolved by position; schema.name-mapping.default is ignored (wrong results)
Key: IMPALA-15444
URL: https://issues.apache.org/jira/browse/IMPALA-15444
Project: IMPALA
Issue Type: Bug
Components: Backend
Reporter: Zoltán Borók-Nagy
h3. Problem
Iceberg tables built by importing existing Parquet or ORC files ({{add_files}},
{{snapshot}}, {{migrate}}) contain data files without field ids. Iceberg's
procedures set {{schema.name-mapping.default}} for them, and the spec ("Column
Projection") requires readers to use that mapping for files without field ids.
Impala ignores the mapping: it numbers the file's columns by their physical
position (the IMPALA-11034 logic for migrated tables) and matches the table's
field ids against those numbers. (Tables created by {{snapshot}} keep their
files outside the table location; with the default
{{-iceberg_allow_datafiles_in_table_location_only=true}}, Impala refuses to
load them before reaching this bug, see IMPALA-11618.)
* If an imported file stores its columns in a different order than the table,
has extra columns, or lacks some of the table's columns, and the types fit,
Impala *silently returns the values of the wrong column*. Nested struct fields
are affected the same way. Row-group min/max and dictionary filtering are
evaluated on the wrong column too.
* If the types do not fit, the query fails with {{File '...' has an
incompatible Parquet schema for column '...'}}.
* {{SET PARQUET_FALLBACK_SCHEMA_RESOLUTION=NAME}} has no effect: the resolver
always uses FIELD_ID for Iceberg tables.
Spark 4.1.3 + Iceberg 1.11.0 reads the same tables correctly. Trino 483 and
ClickHouse 26.8 return the correct results on the benchmark's name-mapping
table.
h3. Steps to reproduce
1. Write a Parquet file without field ids whose column order differs from the
table:
{code:python}
import pyarrow as pa, pyarrow.parquet as pq
pq.write_table(pa.table({"b": pa.array([20, 21], pa.int32()), "a":
pa.array([10, 11], pa.int32())}), "swap_b_a.parquet")
{code}
2. Import it with Spark + Iceberg ({{add_files}} sets
{{schema.name-mapping.default}} when it is missing; it registers files in
place, so copy the file under the table location first):
{code:sql}
CREATE TABLE cat.db.swap (a INT, b INT) USING iceberg TBLPROPERTIES
('format-version'='2');
-- copy swap_b_a.parquet to <table location>/data/imported/
CALL cat.system.add_files(table => 'db.swap', source_table =>
'`parquet`.`<table location>/data/imported`');
INSERT INTO cat.db.swap VALUES (1, 2); -- a natively written file with field
ids
{code}
The table now has {{schema.name-mapping.default =
[{"field-id":1,"names":["a"]},{"field-id":2,"names":["b"]}]}}. (Also reproduced
through the Java API: set {{schema.name-mapping.default}} to
{{MappingUtil.create(schema)}}, compute metrics with
{{ParquetUtil.fileMetrics(file, metricsConfig, nameMapping)}} and append with
{{AppendFiles}}; {{AppendFiles}} does not set the mapping itself. Same result.)
3. Query in Impala (the table was registered as an external
{{'iceberg.catalog'='hadoop.tables'}} table):
{code:sql}
SELECT a, b FROM swap ORDER BY a;
SELECT count(*) FROM swap WHERE a = 10;
SELECT sum(a), sum(b) FROM swap;
{code}
h3. Expected result (Spark returns this)
Rows (1, 2), (10, 20), (11, 21); {{count(*) WHERE a = 10}} = 1; {{sum(a),
sum(b)}} = 22, 43.
h3. Actual result (Impala 4.5.2)
Rows (1, 2), *(20, 10), (21, 11)*; {{count(*) WHERE a = 10}} = *0* (and {{WHERE
a = 20}} = 1); {{sum(a), sum(b)}} = *42, 23*. The natively written row is
correct; in the imported rows {{a}} and {{b}} are swapped. With a file {{(s
STRING, id BIGINT)}} imported into a table {{(id BIGINT, s STRING)}} the query
fails instead:
{noformat}
File '.../err/data/imported/err_s_id.parquet' has an incompatible Parquet
schema for column 'db.err.id'. Column type: BIGINT, Parquet schema:
optional byte_array s [i:0 d:1 r:0]
{noformat}
Master 3d4d47dc30 (5.0.0-SNAPSHOT) gives identical rows, counts, sums and
error. For {{WHERE a = 10}} min/max statistics eliminate the imported file's
row group (RowsRead 0). A file with only column {{b}} imported into {{(a, b)}}
returns {{a}} = 20, 21 and {{b}} = NULL (Spark: {{a}} = NULL, {{b}} = 20, 21).
If the mapping is removed from the table ({{ALTER TABLE ... UNSET TBLPROPERTIES
('schema.name-mapping.default')}}), Spark returns the same swapped rows as
Impala.
h3. Analysis (same logic on 4.5.2 and master 3d4d47dc30; repro confirmed on
master)
* {{ParquetSchemaResolver::Init()}}
({{be/src/exec/parquet/parquet-metadata-utils.h:156-171}}) forces the FIELD_ID
strategy for Iceberg tables and, if the first column has no field id, calls
{{GenerateFieldIDs()}} ({{parquet-metadata-utils.cc:983}} on 4.5.2, {{:1027}}
on master), which assigns 1, 2, 3, ... in physical order.
* Column readers, dictionary filtering and row-group/page statistics all
resolve through the same {{ResolvePath()}}, so values and filters use the wrong
column.
* Nothing in {{be/}} or {{fe/}} reads {{schema.name-mapping.default}}.
* The ORC scanner does the same ({{be/src/exec/orc/orc-metadata-utils.cc:61}}:
{{OrcSchemaResolver::Init()}} calls {{GenerateFieldIDs()}}, defined at {{:330}}
on 4.5.2, {{:334}} on master). Verified on master: an ORC file (b, a) imported
with {{add_files}} returns (20, 10), (21, 11), and {{SET
ORC_SCHEMA_RESOLUTION=NAME}} has no effect.
* Iceberg Java applies position-based fallback ids only when the table has *no*
name mapping ({{ReadConf}}: file ids -> name mapping -> {{addFallbackIds}}).
* IMPALA-11034's description proposed a NameMapping; the commit assigns ids by
position instead. That matches Iceberg's fallback only for tables without a
name mapping. It is wrong whenever a mapping exists and a file's columns differ
from the table's order: imported files, and also in-place migrated Hive
directories, since Hive reads Parquet by name. Impala's own {{ALTER TABLE ...
CONVERT TO ICEBERG}} sets no name mapping ({{MigrateTableUtil}} passes null),
so tables it converts stay position-resolved in every engine, and this fix
would not change them.
* History (code reading): at 4.1.0, Parquet files without field ids were
resolved by NAME (IMPALA-11053, c10e951bcb). That handled reordered, extra and
missing columns but not renames. IMPALA-11034 (4.2.0, backported to 4.1.1)
replaced it with position-based ids. Earlier, IMPALA-10974 fell back to
{{PARQUET_FALLBACK_SCHEMA_RESOLUTION}} (default POSITION). ORC has used
POSITION for such files since IMPALA-10974 (4.1.0).
Suggested fix: pass the table's name mapping (including nested fields and all
aliases) to the Parquet and ORC scanners. For files without field ids, assign
ids by name through it. Columns without a mapped file column follow the spec's
missing-field rules: the identity-partition value (existing
IMPALA-11053/IMPALA-13364 logic), then initial-default (IMPALA-14589), else
NULL. Keep {{GenerateFieldIDs()}} only for tables without a mapping. Follow-up:
have {{CONVERT TO ICEBERG}} set {{schema.name-mapping.default}}, as Iceberg's
{{migrate}} and {{snapshot}} do.
h3. Workaround
Rewrite the imported files with an engine that applies the name mapping, e.g.
Spark {{CALL cat.system.rewrite_data_files(table => 'db.swap', options =>
map('rewrite-all','true'))}} (verified, also on master: Impala returns the
correct rows afterwards).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
