[
https://issues.apache.org/jira/browse/IMPALA-15445?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zoltán Borók-Nagy updated IMPALA-15445:
---------------------------------------
Labels: Avro correctness impala-iceberg (was: correctness impala-iceberg)
> 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
> Priority: Major
> Labels: Avro, correctness, impala-iceberg
>
> 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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
