Zoltán Borók-Nagy created IMPALA-15461:
------------------------------------------
Summary: 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
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)
