Zoltán Borók-Nagy created IMPALA-15445:
------------------------------------------
Summary: Iceberg: Avro data files with DECIMAL, UUID or (read)
TIMESTAMP columns are silently skipped
Key: IMPALA-15445
URL: https://issues.apache.org/jira/browse/IMPALA-15445
Project: IMPALA
Issue Type: Bug
Components: Backend
Reporter: Zoltán Borók-Nagy
h3. Problem
When Impala's Avro scanner cannot handle a type in an Iceberg Avro data file,
it skips the *whole file* and the query still finishes with status OK. The only
trace is a warning (impala-shell "WARNINGS", profile "Errors:"); an HS2 client
that does not fetch the query log sees no error at all. In an Iceberg table
every data file belongs to the committed snapshot, so this is a silent wrong
result. Three types trigger it, all written as the Iceberg spec prescribes
(Appendix A):
* *decimal(P,S)* is Avro {{fixed}} with {{logicalType: decimal}}. The scanner
does not support {{fixed}} and converts the *whole* file schema while parsing
the header, so every file with any DECIMAL column is skipped, even for queries
that never reference that column ({{SELECT id}}, {{count(id)}}).
* *timestamp / timestamptz* is Avro {{long}} with {{logicalType:
timestamp-micros}}. The reader schema Impala derives for the table maps
TIMESTAMP to Avro {{string}}, so schema resolution fails and the file is
skipped whenever the timestamp column is materialized, including when it is
referenced only in a predicate (e.g. {{WHERE ts IS NULL}} returns 0 rows).
* *uuid* (master only; 4.5.2 does not support the Iceberg UUID type) is Avro
{{fixed(16)}} with {{logicalType: uuid}}. Queries that read the UUID column are
rejected at planning time, but a query that does not read it ({{SELECT id}},
{{count(id)}}) skips the file silently, exactly like DECIMAL.
{{count(*)}} is answered from Iceberg metadata without scanning (the snapshot
summary {{total-records}}; for tables with delete files, data-file record
counts), so the same table reports {{count(*) = 3}} and {{count(id) = 0}}. In a
table that also has Parquet files, only the Parquet rows are returned.
h3. Steps to reproduce
Spark 4.1.3 + Iceberg 1.11.0:
{code:sql}
CREATE TABLE avro_dec (id BIGINT, i INT, s STRING, amount DECIMAL(9,2))
USING iceberg TBLPROPERTIES ('format-version'='2',
'write.format.default'='avro');
INSERT INTO avro_dec VALUES (1, 10, 'a', 1.25), (2, 20, 'b', -2.50), (3, NULL,
NULL, NULL);
CREATE TABLE avro_ts (id BIGINT, s STRING, ts TIMESTAMP_NTZ)
USING iceberg TBLPROPERTIES ('format-version'='2',
'write.format.default'='avro');
INSERT INTO avro_ts VALUES (1, 'a', TIMESTAMP_NTZ '2024-02-29
16:27:43.582994'), (2, 'b', NULL);
-- mixed formats: 2 Parquet files, then 2 Avro files
CREATE TABLE t_mixed (id BIGINT, amount DECIMAL(9,2)) USING iceberg
TBLPROPERTIES ('format-version'='2', 'write.format.default'='parquet');
INSERT INTO t_mixed VALUES (1, 1.25), (2, -2.50);
ALTER TABLE t_mixed SET TBLPROPERTIES ('write.format.default'='avro');
INSERT INTO t_mixed VALUES (3, 3.75), (4, 4.00);
{code}
Impala (tables registered as external {{'iceberg.catalog'='hadoop.tables'}}
tables):
{code:sql}
SELECT * FROM avro_dec ORDER BY id;
SELECT count(*), count(id), sum(i) FROM avro_dec;
SELECT * FROM avro_ts ORDER BY id;
SELECT count(*), count(id), sum(amount) FROM t_mixed;
{code}
h3. Expected result (Spark returns this)
{{avro_dec}}: 3 rows, {{count(*), count(id), sum(i)}} = 3, 3, 30. {{avro_ts}}:
(1, 'a', 2024-02-29 16:27:43.582994), (2, 'b', NULL). {{t_mixed}}: 4, 4, 6.50.
If a type cannot be read, the query must fail (as with {{UUID}} on master:
"Reading UUID columns from AVRO format is not yet supported.") and must never
return a partial result with status OK. That guard is incomplete: {{SELECT id}}
on an Avro table with a UUID column returns 0 rows with "Got type: fixed" and
status OK (see *uuid* above).
h3. Actual result (Impala 4.5.2; identical on master 3d4d47dc30)
{noformat}
SELECT * FROM avro_dec ORDER BY id -> 0 rows
WARNINGS: Avro enum, array, map, union, and fixed types are not supported.
Got type: fixed (one warning per Avro data file)
SELECT count(*), count(id), sum(i) FROM avro_dec -> 3 | 0 | NULL (same
warnings)
SELECT * FROM avro_ts ORDER BY id -> 0 rows
WARNINGS: Unresolvable types for column 'ts': table type: string, file type:
long (1 of 2 similar)
SELECT id, s FROM avro_ts ORDER BY id -> correct (ts not read)
SELECT count(*), count(id), sum(amount) FROM t_mixed -> 4 | 2 | -1.25
{noformat}
Profile: {{Query Status: OK}}, {{File Formats: AVRO/NONE(Skipped):<n>}} (one
per skipped file, reported per scan instance), {{RowsRead: 0}}. Neither warning
names the skipped file. With {{SET ABORT_ON_ERROR=true}} the same queries fail
with the warning text.
Master: avro_dec count query {{334cc2f2ca1b59b4:8f745c2c00000000}} -> 3 | 0 |
NULL, t_mixed -> 4 | 2 | -1.25, avro_ts {{SELECT *}} -> 0 rows.
h3. Analysis (4.5.2 and master 3d4d47dc30)
* {{HdfsAvroScanner::ParseMetadata()}} converts the complete writer schema
({{be/src/exec/avro/hdfs-avro-scanner.cc:173}}); {{IsSupportedAvroType()}}
({{be/src/util/avro-util.cc:29-46}}) has no {{AVRO_FIXED}} case. The bundled
avro-c (1.7.4-p5) maps {{logicalType: decimal}} to {{AVRO_DECIMAL}} only for
{{bytes}}.
* {{IcebergTable.setAvroSchema()}} (catalog v1) and {{LocalIcebergTable}} ->
{{LocalFsTable.setAvroSchema()}} (local catalog) derive the Avro reader schema
from the table's Hive-style column list with
{{AvroSchemaConverter.convertFieldSchemas()}}, which maps TIMESTAMP to
{{string}}
({{fe/src/main/java/org/apache/impala/util/AvroSchemaConverter.java:158}}). On
master that column list is the Iceberg schema converted to Impala types (since
IMPALA-13718); in 4.5.2 it was the raw HMS columns. {{VerifyTypesMatch()}}
({{hdfs-avro-scanner.cc:273}}, error at {{:404}}; {{:268}}/{{:399}} in 4.5.2)
then rejects the file's {{long}} timestamp-micros.
* {{BaseSequenceScanner::GetNextInternal()}}
({{be/src/exec/base-sequence-scanner.cc:161-172}} on master, {{:157-168}} in
4.5.2) turns a header error into {{LogOrReturnError()}} (a warning unless
{{ABORT_ON_ERROR}}), and {{CloseFileRanges()}} marks all splits of the file as
skipped. Skipping the whole file on a header error dates back to 2013
(0385d14d69); IMPALA-767 (Impala 1.3, "...Incorrectly Returns 0 Rows") only
made the skip log a warning.
* Master only adds a planning-time guard for UUID
({{FeIcebergTable.validateUuidReadSupported()}},
{{FeIcebergTable.java:178-189}}, called from
{{HdfsScanNode.checkUuidReadSupported()}}, {{HdfsScanNode.java:690-704}},
IMPALA-15200). It fires only when a UUID slot is materialized, so it misses
queries that skip the UUID column. There is nothing for DECIMAL or TIMESTAMP.
Fix directions: for Iceberg tables never skip a data file on a header/schema
error (fail the query regardless of {{ABORT_ON_ERROR}}), or reject at planning
time like the UUID guard; make the UUID guard (or its replacement) check the
file's schema, not only materialized slots; decode fixed-backed decimals
(IMPALA-11876) and {{long}} timestamp-micros; build the Avro reader schema
directly from the Iceberg schema, not through Impala types +
{{AvroSchemaConverter}}; validate only the file fields that are read.
h3. Workaround
{{SET ABORT_ON_ERROR=true}} turns the silent wrong result into an error. For
tables whose only affected type is TIMESTAMP, queries that do not reference the
timestamp column (in SELECT or WHERE) are correct. Otherwise rewrite the data
as Parquet or ORC with another engine, e.g. in Spark {{ALTER TABLE ... SET
TBLPROPERTIES ('write.format.default'='parquet')}} + {{CALL
<catalog>.system.rewrite_data_files(table => '...', options =>
map('rewrite-all', 'true'))}} (verified on master).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
