[
https://issues.apache.org/jira/browse/IMPALA-15461?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zoltán Borók-Nagy resolved IMPALA-15461.
----------------------------------------
Fix Version/s: Impala 5.0.0
Resolution: Fixed
> Iceberg: partition specs are looked up by list position instead of spec id;
> SELECT crashes impalad after expire_snapshots with clean_expired_metadata
> removes an old spec
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: IMPALA-15461
> URL: https://issues.apache.org/jira/browse/IMPALA-15461
> Project: IMPALA
> Issue Type: Bug
> Components: Backend, Frontend
> Reporter: Zoltán Borók-Nagy
> Assignee: Zoltán Borók-Nagy
> Priority: Major
> Labels: impala-iceberg
> Fix For: Impala 5.0.0
>
>
> h3. Problem
> Impala assumes that the partition spec with id N is at position N of the
> table's spec list. The Iceberg spec does not promise this.
> {{partition-specs}} is a list of spec objects, and a manifest's
> {{partition_spec_id}} "must be listed in table metadata {{partition-specs}}",
> i.e. specs are found by id. Since Iceberg 1.8.0,
> {{ExpireSnapshots.cleanExpiredMetadata(true)}} removes every spec that is not
> the default spec and is not used by a manifest of a retained snapshot (unused
> schemas too, since 1.9.0). Spark exposes it as {{CALL ...
> expire_snapshots(..., clean_expired_metadata => true)}} since Iceberg 1.10.0.
> After that a table can have, for example, only spec 1 with
> {{default-spec-id}} 1. This needs no unusual history: it happens to the first
> spec of a table that was re-partitioned before its first write (shorter repro
> below), or to an old spec whose files were all rewritten by compaction (main
> recipe).
> * Backend: {{HdfsTableDescriptor}} reads
> {{iceberg_partition_specs_\[default_partition_spec_id\]}}. With specs \[1\]
> this reads past the end of the vector. This is undefined behavior. In our run
> (master minicluster, query c34d48ebf0562378:5c585f8600000000) both impalads
> that ran a fragment crashed with SIGSEGV while building the descriptor table;
> one was the coordinator. On docker (master 3d4d47dc30, one impalad), 49 of 52
> statements that reached the backend crashed impalad in the same constructor.
> The other 3 returned the correct rows; none returned wrong rows. Most crashes
> are SIGSEGV; some are an abort on {{std::bad_alloc}} or
> {{std::length_error}}, thrown while copying the garbage fields.
> * Frontend (verified): {{getDefaultPartitionSpec()}} does
> {{specs.get(defaultSpecId)}} after {{checkState(specs.size() >
> defaultSpecId)}}. It throws {{IllegalStateException}} for SHOW CREATE TABLE,
> SHOW PARTITIONS, CREATE TABLE LIKE, INSERT, INSERT OVERWRITE, UPDATE, MERGE
> and OPTIMIZE. It also throws when the scan of the table has a WHERE predicate
> on a table column or gets a runtime filter from a join: SELECT ... WHERE,
> joins, and DELETE (a DELETE without WHERE is rejected with the TRUNCATE
> hint). Everything else that scans the table reaches the backend and crashes
> impalad (verified): SELECT, SELECT count(id), queries on its metadata tables
> (e.g. {{t.snapshots}}), time travel, COMPUTE STATS, INSERT ... SELECT and
> CTAS from the table, and joins or DELETE ... WHERE id IN (subquery) with
> RUNTIME_FILTER_MODE=OFF. SELECT count(*) is answered from snapshot metadata
> and does not crash.
> * If the default id is smaller than the list size, it silently returns
> another spec: with specs \[1, 2\] and default 1 (spec 1 unpartitioned, spec 2
> identity(region)), SHOW CREATE TABLE prints {{PARTITIONED BY SPEC (region)}}
> and INSERT is planned with {{PARTITION-KEYS}}.
> Spark reads these tables fine.
> h3. Steps to reproduce
> Verified on master 3d4d47dc30 (RELEASE, docker, REST catalog, one impalad):
> specs \[1\] with default 1, Spark returns 1, 2, 3, EXPLAIN of the SELECT
> plans normally (7846550d1fb08eb5:351d9e0400000000), and the SELECT crashes
> impalad (f94305d4fa598f69:b94bc09500000000; 19 of 20 runs on two fresh copies
> of the table crashed, 1 returned the correct rows). The {{rw_b}} variant
> (below) crashed impalad on both the minicluster and docker.
> Spark 4.1.3 + Iceberg 1.11.0 ({{ath}} is the Spark catalog name of the
> Iceberg REST catalog):
> {code:sql}
> CREATE NAMESPACE IF NOT EXISTS ath.zz_jira;
> CREATE TABLE ath.zz_jira.nondense (id BIGINT, region STRING) USING iceberg
> PARTITIONED BY (region) TBLPROPERTIES ('format-version'='2');
> INSERT INTO ath.zz_jira.nondense VALUES (1, 'EU'), (2, 'NA'); -- spec 0 =
> identity(region)
> ALTER TABLE ath.zz_jira.nondense DROP PARTITION FIELD region; -- spec 1
> (no fields), now default
> CALL ath.system.rewrite_data_files(table => 'zz_jira.nondense', options =>
> map('rewrite-all', 'true'));
> INSERT INTO ath.zz_jira.nondense VALUES (3, 'AS'); -- drops the
> spec-0 manifest that has only DELETED entries
> CALL ath.system.expire_snapshots(table => 'zz_jira.nondense', older_than =>
> current_timestamp(),
> retain_last => 1, clean_expired_metadata => true);
> -- metadata: "partition-specs" : [ {"spec-id" : 1, "fields" : [ ]} ],
> "default-spec-id" : 1
> SELECT * FROM ath.zz_jira.nondense ORDER BY id; -- Spark: 1,
> 2, 3
> {code}
> Impala sees the table through the REST catalog (impalad
> {{--catalog_config_dir}}), or after {{CREATE EXTERNAL TABLE ... STORED AS
> ICEBERG LOCATION '...' TBLPROPERTIES('iceberg.catalog'='hadoop.tables')}}
> (the location needs the HadoopTables layout: {{metadata/vN.metadata.json}}
> and {{version-hint.text}}). Then: {{SELECT * FROM zz_jira.nondense ORDER BY
> id;}}
> Without the INSERT between rewrite_data_files and expire_snapshots, spec 0
> survives: the only retained snapshot still has a spec-0 manifest with DELETED
> entries (verified with this recipe: specs \[0, 1\] after expire; Impala reads
> it, query 42462edf832efaf4:e40575ad00000000).
> Silent wrong spec: after the recipe, {{ALTER TABLE ... ADD PARTITION FIELD
> region; ALTER TABLE ... DROP PARTITION FIELD region;}} gives specs \[1, 2\]
> with default 1 (Iceberg reuses spec id 1).
> Crash variant ({{rw_b}}, 2026-09-28, master minicluster, hadoop.tables):
> differs by an extra {{INSERT (3, 'AS')}} after DROP PARTITION FIELD, {{DROP
> COLUMN region}} before the rewrite, {{INSERT VALUES (4)}} as the last insert,
> and expiration through the Java API ({{cleanExpiredMetadata(true)}}; on
> docker the Spark procedure, same result). Result: specs \[1\], schemas \[1\],
> Spark returns 1-4.
> Shorter repro (verified: specs \[1\] with default 1, Spark returns 1, 2, 3,
> the Impala SELECT crashes impalad, 4b42f3feea4ed23f:d53da1a500000000). The
> old spec never held data, so no rewrite is needed, and the expiration removes
> no snapshot:
> {code:sql}
> CREATE TABLE ath.zz_jira.mini (id BIGINT, region STRING) USING iceberg
> PARTITIONED BY (region) TBLPROPERTIES ('format-version'='2');
> ALTER TABLE ath.zz_jira.mini DROP PARTITION FIELD region;
> INSERT INTO ath.zz_jira.mini VALUES (1, 'EU'), (2, 'NA'), (3, 'AS');
> CALL ath.system.expire_snapshots(table => 'zz_jira.mini', older_than =>
> current_timestamp(),
> retain_last => 1, clean_expired_metadata => true);
> {code}
> h3. Expected result
> The rows Spark returns (1, 2, 3).
> h3. Actual result (master 3d4d47dc30, RELEASE)
> Minicluster: {{SELECT * FROM rw_b ORDER BY id}} (query
> c34d48ebf0562378:5c585f8600000000): impala-shell gets {{TSocket read 0
> bytes}}. The coordinator (pid 2049916) and an executor (pid 2049918) crashed.
> Both cores have the same stack:
> {noformat}
> #11 __memmove_avx_unaligned_erms
> #18 impala::TIcebergPartitionField::TIcebergPartitionField
> (CatalogObjects_types.cpp:3310)
> #22 std::vector<impala::TIcebergPartitionField>::push_back
> #23 impala::HdfsTableDescriptor::HdfsTableDescriptor (descriptors.cc:294)
> #24 impala::DescriptorTbl::CreateTblDescriptorInternal (descriptors.cc:637)
> #25 impala::DescriptorTbl::CreateInternal (descriptors.cc:676)
> #26 impala::DescriptorTbl::Create (descriptors.cc:667)
> #27 impala::QueryState::StartFInstances (query-state.h:166)
> #28 impala::QueryExecMgr::ExecuteQueryHelper (query-exec-mgr.cc:173)
> {noformat}
> In frame 23 of core.2049916 the table is {{athl_v5x.rw_b}},
> {{iceberg_spec_id_}} = 1, and {{iceberg_partition_specs_}} has one element,
> with {{spec_id}} 1 and 0 fields.
> Frontend results on docker (EXPLAIN does not start fragments; REST table with
> specs \[1\], DML on the same table registered in HMS as hadoop.tables):
> ||Statement||Result||Query id||
> |EXPLAIN SELECT * ORDER BY id|plans
> normally|7846550d1fb08eb5:351d9e0400000000|
> |EXPLAIN SELECT ... WHERE id =
> 1|IllegalStateException|764ae9ab0790078b:959b730d00000000|
> |EXPLAIN self-join on
> region|IllegalStateException|5f484b2d1ac66464:2c16331d00000000|
> |SHOW CREATE TABLE / SHOW
> PARTITIONS|IllegalStateException|d5411b033751c2a6:7bf590fc00000000,
> e6468807f7404e96:c431a07200000000|
> |EXPLAIN INSERT / INSERT
> OVERWRITE|IllegalStateException|8e4343456525cd74:d6522cd900000000,
> e043fe82b2c02021:0ea9696f00000000|
> |EXPLAIN DELETE ... WHERE /
> UPDATE|IllegalStateException|084e193874245e9e:bd7cbec000000000,
> 87490281d32e1c20:6d8011e500000000|
> |specs \[1, 2\], default 1: SHOW CREATE TABLE|PARTITIONED BY SPEC
> (region)|a14502f29d422b4b:4d8451f200000000|
> |specs \[1, 2\], default 1: EXPLAIN
> INSERT|PARTITION-KEYS=('xx')|36418e2603cf1ca6:cb14990000000000|
> The control table with specs \[0, 1\] runs the REST statements normally (SHOW
> PARTITIONS answers "Table is not partitioned", as expected for an
> unpartitioned default spec).
> h3. Analysis (master 3d4d47dc30)
> * The frontend sends the specs as a list, in map order, not indexed by id.
> {{FeIcebergTable.Utils.loadPartitionSpecByIceberg()}}
> ({{FeIcebergTable.java:1218-1226}}) iterates {{specs().values()}}.
> {{getTIcebergTable()}} ({{:974-978}}) adds them to
> {{TIcebergTable.partition_spec}} and sets {{default_partition_spec_id}}
> ({{common/thrift/CatalogObjects.thrift:717-718}}; the id of each spec is
> {{TIcebergPartitionSpec.spec_id}}, {{:654}}).
> * Backend, {{be/src/runtime/descriptors.cc:287-295}}: {{iceberg_spec_id_ =
> default_partition_spec_id}}, then {{const TIcebergPartitionSpec& spec =
> iceberg_partition_specs_\[iceberg_spec_id_\]}} (:289). The
> {{DCHECK_EQ(spec.spec_id, iceberg_spec_id_)}} (:290) exists only in debug
> builds. The loop at :291-295 then copies the fields of whatever memory
> follows the vector.
> * Backend, same assumption (verified):
> {{IcebergDeleteSinkBase::ConstructPartitionInfo()}} uses
> {{IcebergPartitionSpecs()\[spec_id\]}}
> ({{be/src/exec/iceberg-delete-sink-base.cc:152-153}}) for position deletes of
> data files written with a non-default spec (DELETE, UPDATE, MERGE). With
> specs \[1\] these statements fail before the sink. With specs \[1, 2\],
> default 1, and a data file written with spec 2 (after the recipe: ADD
> PARTITION FIELD region, INSERT, DROP PARTITION FIELD region), a DELETE of a
> row in that file reads index 2 of a 2-element vector and crashed impalad in
> {{ConstructPartitionInfo}} ({{iceberg-delete-sink-base.cc:157}}, from
> {{IcebergBufferedDeleteSink::FlushBufferedRecords}}):
> b64c34605120c14c:bd7667d500000000 (SIGSEGV),
> d24bdeff14213c0e:5e7a09f300000000 (abort on std::bad_alloc).
> * Backend, metadata tables and time travel (verified):
> {{IcebergMetadataTable.toThriftDescriptor()}}
> ({{IcebergMetadataTable.java:123-126}}) sends the base table's specs in an
> ICEBERG_TABLE descriptor, so {{CreateTblDescriptorInternal}}
> ({{descriptors.cc:635-637}}) builds an {{HdfsTableDescriptor}} for it too.
> {{SELECT ... FROM t.snapshots}} and {{t.`files`}} crashed impalad in 3 of 4
> runs (5348d780119d84f1:f3f8a9b000000000), and so did a time-travel SELECT
> (9740a78b28f9e22c:96f96ae600000000).
> * Frontend: {{FeIcebergTable.Utils.getDefaultPartitionSpec()}}
> ({{FeIcebergTable.java:1243-1251}}) throws at {{:1249}}. Verified with the
> stacks the impalad log has at glog level 1 (the default level logs no stack):
> SELECT ... WHERE and DELETE ... WHERE via
> {{IcebergScanPlanner.java:1020/1048/1227}}, joins via
> {{RuntimeFilterGenerator.java:1505}} ->
> {{FeIcebergTable.isComputedPartitionColumn()}} (:423), SHOW CREATE TABLE via
> {{ToSqlUtils.java:564}}, SHOW PARTITIONS via {{ShowStatsStmt.java:142}},
> INSERT via {{InsertStmt.java:868}}, INSERT OVERWRITE via
> {{InsertStmt.java:653}}, UPDATE via {{IcebergUpdateImpl.java:118}}, MERGE via
> {{IcebergUtil.populatePartitionExprs()}} ({{IcebergUtil.java:1793}}) from
> {{IcebergMergeImpl.java:176}}. By code reading, OPTIMIZE
> ({{OptimizeStmt.java:297}}) and UPDATE ({{IcebergUpdateImpl.java:141}}) also
> call {{populatePartitionExprs()}}, CREATE TABLE LIKE calls
> {{getDefaultPartitionSpec()}} in catalogd ({{CatalogOpExecutor.java:4693}}),
> and {{InsertStmt.java:534}} is another caller.
> * A lookup by id already exists: {{FeIcebergTable.getPartitionSpec(int
> specId)}} ({{:233-238}}), used in
> {{IcebergCatalogOpExecutor.java:587/631/652}}.
> * History: the frontend lookup and the {{default_partition_spec_id}} field
> came with IMPALA-10184 (a5019eb12e, 4.0.0). The lookup in
> {{HdfsTableDescriptor}} came with IMPALA-10380 (296ed74d6f, 4.0.0).
> IMPALA-11877 part 1 (c579212251, 4.3.0) added the DCHECK_EQ. Part 2
> (ce4202a70b, 4.3.0) moved the lookup into {{iceberg_partition_specs_}} and
> added the delete sink lookup, which IMPALA-12313 part 3 (68fe57ff84, 4.4.0)
> moved to {{iceberg-delete-sink-base.cc}}. So every release since 4.0.0 has
> the descriptor lookup (code reading).
> * 4.5.2 has the same code ({{descriptors.cc:261}},
> {{iceberg-delete-sink-base.cc:151}}, {{FeIcebergTable.java:853-854}}) and the
> same behavior. Verified on the apache/impala:4.5.2 images (docker,
> hadoop.tables): the same IllegalStateExceptions (e.g.
> 614ca16fde04c17c:3737c23d00000000, a64ff5b4fbfdee2e:52076d1900000000) and the
> wrong spec for specs \[1, 2\] with default 1. SELECT crashed impalad in 7 of
> 8 runs (05467358c27c9ba6:f592a26700000000) and a metadata-table query in 2 of
> 4 runs (0b4366fabd375810:3c268cff00000000), in
> {{HdfsTableDescriptor::HdfsTableDescriptor}}.
> * Impala itself does not create such tables. Its {{ALTER TABLE ... EXECUTE
> EXPIRE_SNAPSHOTS}} never calls {{cleanExpiredMetadata}}
> ({{IcebergCatalogOpExecutor.java:258-266}}). Other writers can (API since
> Iceberg 1.8.0, Spark procedure since 1.10.0; checked in the Iceberg sources
> and jars); the removal is the {{MetadataUpdate.RemovePartitionSpecs}} table
> update.
> Suggested fix: look specs up by id everywhere. In the backend, build an id ->
> spec map in the {{HdfsTableDescriptor}} constructor and return an error if
> the default id is missing. The constructor cannot return a Status, so check
> in {{DescriptorTbl::CreateTblDescriptorInternal}}
> ({{descriptors.cc:635-640}}) or add an Init(). Use the map in
> {{iceberg-delete-sink-base.cc}} too. In the frontend, implement
> {{getDefaultPartitionSpec()}} as
> {{getPartitionSpec(getDefaultPartitionSpecId())}}. Keep {{partition_spec}} a
> list: {{TIcebergTable}} is shared by catalogd and impalad, and the thrift
> compatibility check (IMPALA-13305) flags a changed field type. Tests: a table
> without spec 0 (SELECT, INSERT, DELETE, SHOW CREATE TABLE), and specs \[1,
> 2\] with default 1 (ADD PARTITION FIELD then DROP PARTITION FIELD after the
> recipe; verified).
> h3. Workaround
> Do not run {{expire_snapshots}} with {{clean_expired_metadata}} on tables
> that Impala reads. An affected table has to be copied into a new table (e.g.
> Spark CTAS), which starts again at spec 0. Verified: the Spark CTAS copy has
> specs \[0\] and Impala returns 1, 2, 3 (f04b9db466b0b9e6:d4fab6a200000000).
> Make the copy with another engine: an Impala CTAS or INSERT ... SELECT reads
> the affected table and crashes impalad (verified: INSERT ... SELECT
> bc4657ff539c87b7:c4c3888500000000, CTAS 1e4a8ec78e0c9d32:3311ae5b00000000,
> ff4e09ecfd378505:7e983bf800000000).
> Adding a new spec does not help: its id is 2, and \[1, 2\] with default 2 is
> still out of range (verified: same IllegalStateExceptions,
> a440a1878dd1d24c:4fc8b31600000000, 3b4497ad5c3dcf28:f064050d00000000; EXPLAIN
> SELECT * plans normally, 49477f4e20dca99c:b9398b7d00000000). Going back to
> the old spec reuses id 1, and Impala silently uses spec 2.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
