From 1f71a665654f60ac5895440a3c86e02ea8a36825 Mon Sep 17 00:00:00 2001 From: "Mingyu Chen (Rayner)" Date: Tue, 3 Nov 2026 10:27:49 +0000 Subject: [PATCH] [fix](fe) Keep sibling capabilities on iceberg-on-HMS tables pinned at a snapshot (#69999) ### What problem does this PR solve? Issue Number: None Related PR: #69299, #69488 Problem Summary: #### In short An Iceberg table stored in HMS and read through an HMS catalog (`'type'='hms'`) loses every capability it inherits from the embedded Iceberg connector in any statement that pins the table at a snapshot. Since #69299 that is every statement that reads it: it gets no Top-N lazy materialization and no nested column pruning, and `DEFAULT(column)` in the conditions of `UPDATE` and `MERGE` silently matches nothing. This PR makes the snapshot overload of `HiveConnectorMetadata#getTableSchema` pass the inherited capabilities on, as the latest-schema overload already does. It is a master-only fix (branch-4.x has no connector SPI), left over from #69488 (review comment 4355508100). #### Background fe-core resolves a table-scoped capability (`PluginDrivenExternalTable.hasCapability`) from the catalog's connector or from the per-table set in the table's schema. It never asks the embedded connector that built the schema of an iceberg-on-HMS table. So `HiveConnectorMetadata` copies the capabilities that fe-core resolves per table (`SIBLING_INHERITABLE_CAPABILITIES`: auto and sampled analyze, Top-N lazy materialization, nested column pruning, field-id access paths, storage predicate pruning, nested column DDL and write-only defaults) from the owning connector into the per-table set of each schema it delegates. Iceberg declares seven of the eight; it does not declare sampled analyze. It did this only in `getTableSchema(session, handle)`. `getTableSchema(session, handle, snapshot)` returned the sibling's schema unchanged. fe-core uses that overload for explicit time travel. Since #69299 it also uses it for the latest-snapshot pin of every statement that reads an Iceberg table: the pin retains its schema (`retainSchema(true)`), `PluginDrivenMvccExternalTable` builds the schema at that pin, and `getSchemaCacheValue()`, where `hasCapability` reads the per-table set, returns it for the rest of the statement. #### The problem, and what it cost Reproduced on a local cluster with a standalone Hive 3.1.2 metastore. Two identical format-version 3 tables `(id INT, c INT, s STRUCT)` get two rows, then `ADD COLUMN d INT DEFAULT 42`, `RENAME COLUMN s.a TO a2` and a third row with `d = 7`. The same statements then run against one table through the Iceberg catalog and against the other through the HMS catalog. | Statement | Iceberg catalog | HMS catalog, master | HMS catalog, this PR | |---|---|---|---| | `EXPLAIN SELECT * FROM t ORDER BY id LIMIT 1`, also with `FOR VERSION AS OF` | `VMaterializeNode` | no lazy materialization | `VMaterializeNode` | | `EXPLAIN SELECT struct_element(s, 'a2') FROM t` | `all access paths: [s.a2]` | no pruning | `all access paths: [s.a2]` | | `UPDATE t SET c = 100 WHERE d = DEFAULT(d)` | updates rows 0 and 1 | updates nothing | updates rows 0 and 1 | | `MERGE INTO t USING (SELECT 0 AS id) s ON t.id = s.id AND t.d = DEFAULT(t.d) WHEN MATCHED THEN UPDATE SET c = 300` | updates row 0 | updates nothing | updates row 0 | | `DELETE FROM t WHERE d = DEFAULT(d)` | fails: `cannot be used in a DELETE` | deletes nothing | fails the same way | | `SELECT DEFAULT(d) FROM t` | fails: `must reference a column of the table that this statement writes` | returns NULL | fails the same way | The `UPDATE`, `MERGE` and `DELETE` rows fail silently: the statement succeeds and changes nothing. Reads of the renamed field return the right values in all three columns: on master the HMS catalog does not prune the struct at all, and with this PR it prunes it with field-id access paths, as the Iceberg catalog does. `INSERT ... VALUES (DEFAULT(d))` and `INSERT ... SELECT DEFAULT(d)` already wrote 42 through both catalogs. #### How this PR fixes it Both overloads now go through one private helper, `delegatedTableSchema`. It resolves the owning sibling once, asks it for the schema (at the pin, for the snapshot overload), and passes the result through the existing `reflectSiblingCapabilities`. The snapshot reaches the sibling unchanged. A hudi-on-HMS table is unaffected: the Hudi connector declares no capabilities, so its schema comes back as it was. A plain Hive table never reaches this branch. #### Results Through the HMS catalog, every statement in the table above now plans and answers as it does through the Iceberg catalog. --- .../connector/hive/HiveConnectorMetadata.java | 31 +++-- ...onnectorMetadataSiblingDelegationTest.java | 118 +++++++++++++----- ...erg_on_hms_gateway_pinned_capabilities.out | 39 ++++++ ..._on_hms_gateway_pinned_capabilities.groovy | 118 ++++++++++++++++++ 4 files changed, 269 insertions(+), 37 deletions(-) create mode 100644 regression-test/data/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.out create mode 100644 regression-test/suites/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.groovy diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java index 43ae99f7bb3..700806051df 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java @@ -556,12 +556,8 @@ public class HiveConnectorMetadata implements ConnectorMetadata { // delegated schema as a per-table marker so it survives delegation (mirrors Trino table-redirection, // where the redirected-to connector's capabilities govern the table). Only the subset fe-core // actually resolves per-table is reflected; a catalog-wide capability (view / show-create / mvcc) - // would be discarded by the reader, so emitting it would only record an unhonoured intent. Resolve - // the owner ONCE (getMetadata is not free) and reuse it for the schema build and the capability read. - SiblingOwner owner = siblingOwnerResolver.apply(handle); - ConnectorTableSchema siblingSchema = memoizedSiblingMetadata(session, owner.connector(), owner.label()) - .getTableSchema(session, handle); - return reflectSiblingCapabilities(owner.connector(), siblingSchema); + // would be discarded by the reader, so emitting it would only record an unhonoured intent. + return delegatedTableSchema(session, handle, sibling -> sibling.getTableSchema(session, handle)); } HiveTableHandle hiveHandle = (HiveTableHandle) handle; String dbName = hiveHandle.getDbName(); @@ -651,6 +647,24 @@ public class HiveConnectorMetadata implements ConnectorMetadata { ConnectorCapability.SUPPORTS_NESTED_COLUMN_SCHEMA_CHANGE, ConnectorCapability.SUPPORTS_WRITE_ONLY_DEFAULTS)); + /** + * A delegated (iceberg/hudi-on-HMS) table's schema as {@code buildSchema} asks its owning sibling for it, + * with the capabilities the table inherits from that sibling reflected onto it + * ({@link #reflectSiblingCapabilities}). Both {@code getTableSchema} overloads go through here because + * fe-core reads a table's per-table capabilities from whichever of the two schemas the statement uses: a + * statement that pins the table (explicit time travel, and every read of an iceberg table, whose + * latest-snapshot pin retains its schema) uses the snapshot overload's, so the two must carry the same + * capabilities. The owner is resolved ONCE (getMetadata is not free) and reused for the schema build and the + * capability read. + */ + private ConnectorTableSchema delegatedTableSchema(ConnectorSession session, ConnectorTableHandle handle, + Function buildSchema) { + SiblingOwner owner = siblingOwnerResolver.apply(handle); + ConnectorTableSchema siblingSchema = buildSchema.apply( + memoizedSiblingMetadata(session, owner.connector(), owner.label())); + return reflectSiblingCapabilities(owner.connector(), siblingSchema); + } + /** * Reflects the {@link #SIBLING_INHERITABLE_CAPABILITIES} subset of the owning sibling connector's * connector-wide capability set onto a delegated (iceberg/hudi-on-HMS) table's schema as per-table @@ -1598,7 +1612,10 @@ public class HiveConnectorMetadata implements ConnectorMetadata { public ConnectorTableSchema getTableSchema(ConnectorSession session, ConnectorTableHandle handle, ConnectorMvccSnapshot snapshot) { if (!(handle instanceof HiveTableHandle)) { - return siblingMetadata(session, handle).getTableSchema(session, handle, snapshot); + // The schema a pinned statement reads its per-table capabilities from, so it inherits exactly what + // the latest-schema overload above does. + return delegatedTableSchema(session, handle, + sibling -> sibling.getTableSchema(session, handle, snapshot)); } // Hive has no schema-at-snapshot; the SPI default ignores the snapshot and returns the latest schema. return getTableSchema(session, handle); diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java index 77a7b1529eb..6f43622b053 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java @@ -70,6 +70,7 @@ import java.util.Optional; import java.util.OptionalLong; import java.util.Set; import java.util.TreeSet; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BooleanSupplier; import java.util.function.Supplier; import java.util.regex.Matcher; @@ -223,6 +224,9 @@ public class HiveConnectorMetadataSiblingDelegationTest { elsewhere.put("icebergSiblingMetadata", "the iceberg by-type resolver, driven by HiveConnectorSiblingTest"); elsewhere.put("hudiSiblingMetadata", "the hudi by-type resolver, the other half of the same pair"); elsewhere.put("memoizedSiblingMetadata", "the per-session memo behind the two resolvers above"); + elsewhere.put("delegatedTableSchema", + "the body both getTableSchema overloads share, driven through each of them by the" + + " capability-reflection cases here"); elsewhere.put("getTableHandle", "the pivot that PRODUCES a foreign handle rather than consuming one, so it cannot be driven" + " with the foreignHandle these cases use; covered by HiveConnectorThreeWayRoutingTest"); @@ -512,6 +516,24 @@ public class HiveConnectorMetadataSiblingDelegationTest { Assertions.assertTrue(siblingMetadata.calls.isEmpty(), "the sibling must not be forwarded a hive handle"); } + /** The pin of a statement that reads an iceberg table: its latest snapshot, with the schema retained. */ + private static final ConnectorMvccSnapshot PINNED_SNAPSHOT = + ConnectorMvccSnapshot.builder().snapshotId(7L).schemaId(3L).retainSchema(true).build(); + + /** + * The schema each getTableSchema overload returns for the foreign handle, keyed by the overload. fe-core reads + * a table's per-table capabilities from the snapshot overload's schema in every statement that pins the table + * (explicit time travel, and every read of an iceberg table, whose latest-snapshot pin retains its schema) and + * from the latest-schema overload's otherwise, so what a delegated schema inherits must hold for both. + */ + private Map schemasFromBothOverloads(HiveConnectorMetadata md) { + Map schemas = new LinkedHashMap<>(); + schemas.put("getTableSchema(session, handle)", md.getTableSchema(session, foreignHandle)); + schemas.put("getTableSchema(session, handle, snapshot)", + md.getTableSchema(session, foreignHandle, PINNED_SNAPSHOT)); + return schemas; + } + @Test public void foreignHandleSchemaReflectsSiblingScanCapabilitiesAsPerTableMarker() { // Option C: fe-core's PluginDrivenExternalTable.hasCapability reads only the CATALOG (hive) connector, @@ -519,7 +541,8 @@ public class HiveConnectorMetadataSiblingDelegationTest { // capabilities onto the delegated schema as a per-table marker, or an iceberg-on-HMS table silently loses // auto-analyze / Top-N lazy / nested-column prune / storage predicate pruning (all declared connector-wide). // MUTATION: dropping the reflection -> the returned schema carries no marker -> the embedded table drops the - // capabilities post-flip -> red here. + // capabilities post-flip -> red here. Dropping it from the snapshot overload alone is red too: every + // statement reading an iceberg-on-HMS table would then plan without them. Set siblingCaps = EnumSet.of( ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE, ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE, @@ -532,24 +555,56 @@ public class HiveConnectorMetadataSiblingDelegationTest { handle -> new SiblingOwner(new CapabilityDeclaringSiblingConnector(siblingCaps), SiblingOwner.ICEBERG_LABEL)); - ConnectorTableSchema schema = md.getTableSchema(session, foreignHandle); - Set reflected = schema.getTableCapabilities(); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_WRITE_ONLY_DEFAULTS), - "an iceberg-on-HMS table must inherit write-only defaults, or DEFAULT(column) on it answers " - + "NULL from the read schema"); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE), - "auto-analyze must survive the delegation as a per-table capability"); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE), - "Top-N lazy must survive the delegation as a per-table capability"); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_NESTED_COLUMN_PRUNE), - "nested-column prune must survive the delegation as a per-table capability"); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_FIELD_ID_ACCESS_PATH), - "an iceberg-on-HMS table must inherit the field-id access path capability, or its nested " - + "access paths stay name-based while BE matches them by iceberg field id"); - Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_STORAGE_PREDICATE_PRUNING), - "storage predicate pruning must survive the delegation as a per-table capability"); - Assertions.assertEquals("sibling-generation", schema.getWriteMetadataIdentity(), - "capability reflection must not discard the sibling's write-generation fence"); + for (Map.Entry entry : schemasFromBothOverloads(md).entrySet()) { + String overload = entry.getKey(); + ConnectorTableSchema schema = entry.getValue(); + Set reflected = schema.getTableCapabilities(); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_WRITE_ONLY_DEFAULTS), + overload + ": an iceberg-on-HMS table must inherit write-only defaults, or DEFAULT(column) on " + + "it answers NULL from the read schema"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE), + overload + ": auto-analyze must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE), + overload + ": Top-N lazy must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_NESTED_COLUMN_PRUNE), + overload + ": nested-column prune must survive the delegation as a per-table capability"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_FIELD_ID_ACCESS_PATH), + overload + ": an iceberg-on-HMS table must inherit the field-id access path capability, or " + + "its nested access paths stay name-based while BE matches them by iceberg field id"); + Assertions.assertTrue(reflected.contains(ConnectorCapability.SUPPORTS_STORAGE_PREDICATE_PRUNING), + overload + ": storage predicate pruning must survive the delegation as a per-table capability"); + Assertions.assertEquals("sibling-generation", schema.getWriteMetadataIdentity(), + overload + ": capability reflection must not discard the sibling's write-generation fence"); + } + } + + @Test + public void snapshotOverloadForwardsThePinAndResolvesTheOwnerOnce() { + // The snapshot overload must hand the sibling the statement's own pin (a schema at another snapshot would + // describe other columns) and, like the latest-schema overload, resolve the owning sibling once for both + // the schema build and the capability read. + // MUTATION: asking the sibling for its latest schema instead -> the pin never reaches it -> red; resolving + // the owner a second time for the capability read -> red on the count. + CapabilityDeclaringSiblingConnector sibling = new CapabilityDeclaringSiblingConnector( + EnumSet.of(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE)); + AtomicInteger resolutions = new AtomicInteger(); + HiveConnectorMetadata md = new HiveConnectorMetadata(null, HiveTestProperties.minimal(), new FakeConnectorContext(), + SUPPLIER_MUST_NOT_BE_USED, SUPPLIER_MUST_NOT_BE_USED, + handle -> { + resolutions.incrementAndGet(); + return new SiblingOwner(sibling, SiblingOwner.ICEBERG_LABEL); + }); + + ConnectorTableSchema schema = md.getTableSchema(session, foreignHandle, PINNED_SNAPSHOT); + + Assertions.assertEquals(Collections.singletonList("getTableSchemaAtSnapshot"), sibling.metadata.calls, + "the snapshot overload must ask the sibling for its schema at the pin, not its latest schema"); + Assertions.assertSame(PINNED_SNAPSHOT, sibling.metadata.lastSchemaSnapshot, + "the sibling must receive the statement's pin unchanged"); + Assertions.assertEquals(EnumSet.of(ConnectorCapability.SUPPORTS_TOPN_LAZY_MATERIALIZE), + schema.getTableCapabilities(), "the schema at the pin inherits the sibling's capability"); + Assertions.assertEquals(1, resolutions.get(), + "the owning sibling is resolved once for the schema build and the capability read"); } @Test @@ -570,10 +625,9 @@ public class HiveConnectorMetadataSiblingDelegationTest { handle -> new SiblingOwner(new CapabilityDeclaringSiblingConnector(siblingCaps), SiblingOwner.ICEBERG_LABEL)); - ConnectorTableSchema schema = md.getTableSchema(session, foreignHandle); - Assertions.assertEquals(EnumSet.of(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE), - schema.getTableCapabilities(), - "only the per-table-resolved subset may be reflected onto a delegated table"); + schemasFromBothOverloads(md).forEach((overload, schema) -> Assertions.assertEquals( + EnumSet.of(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE), schema.getTableCapabilities(), + overload + ": only the per-table-resolved subset may be reflected onto a delegated table")); // A NON-empty sibling declaring none of the subset must inherit nothing. HiveConnectorMetadata noneInheritable = new HiveConnectorMetadata(null, HiveTestProperties.minimal(), @@ -581,9 +635,10 @@ public class HiveConnectorMetadataSiblingDelegationTest { handle -> new SiblingOwner(new CapabilityDeclaringSiblingConnector( EnumSet.of(ConnectorCapability.SUPPORTS_VIEW, ConnectorCapability.SUPPORTS_MVCC_SNAPSHOT)), SiblingOwner.ICEBERG_LABEL)); - Assertions.assertTrue(noneInheritable.getTableSchema(session, foreignHandle) - .getTableCapabilities().isEmpty(), - "a sibling declaring only catalog-wide capabilities must contribute no per-table capability"); + schemasFromBothOverloads(noneInheritable).forEach((overload, schema) -> Assertions.assertTrue( + schema.getTableCapabilities().isEmpty(), + overload + ": a sibling declaring only catalog-wide capabilities must contribute no per-table " + + "capability")); } @Test @@ -594,9 +649,9 @@ public class HiveConnectorMetadataSiblingDelegationTest { // foreignHandleSchemaReflectsOnlyThePerTableResolvedCapabilitySubset above. // MUTATION: dropping the isEmpty() early-return and rebuilding the schema unconditionally -> red here. HiveConnectorMetadata md = withSibling(); // RecordingSiblingConnector declares no capabilities - ConnectorTableSchema schema = md.getTableSchema(session, foreignHandle); - Assertions.assertTrue(schema.getTableCapabilities().isEmpty(), - "nothing inherited when the sibling declares no capabilities"); + schemasFromBothOverloads(md).forEach((overload, schema) -> Assertions.assertTrue( + schema.getTableCapabilities().isEmpty(), + overload + ": nothing inherited when the sibling declares no capabilities")); } // ============== per-statement sibling-metadata funnel (HMS heterogeneous gateway) ============== @@ -719,6 +774,7 @@ public class HiveConnectorMetadataSiblingDelegationTest { /** A sibling {@link Connector} declaring a fixed capability set; its metadata returns a marker-less schema. */ private static final class CapabilityDeclaringSiblingConnector implements Connector { private final Set caps; + private final RecordingSiblingMetadata metadata = new RecordingSiblingMetadata(); CapabilityDeclaringSiblingConnector(Set caps) { this.caps = caps; @@ -731,7 +787,7 @@ public class HiveConnectorMetadataSiblingDelegationTest { @Override public ConnectorMetadata getMetadata(ConnectorSession session) { - return new RecordingSiblingMetadata(); + return metadata; } } @@ -772,6 +828,7 @@ public class HiveConnectorMetadataSiblingDelegationTest { "deleteUsingMetadata")); final List calls = new ArrayList<>(); + ConnectorMvccSnapshot lastSchemaSnapshot; ConnectorExpression lastDeletePredicate; String lastBoundWriteMetadataIdentity; BooleanSupplier lastCancellation; @@ -789,6 +846,7 @@ public class HiveConnectorMetadataSiblingDelegationTest { public ConnectorTableSchema getTableSchema(ConnectorSession session, ConnectorTableHandle handle, ConnectorMvccSnapshot snapshot) { calls.add("getTableSchemaAtSnapshot"); + lastSchemaSnapshot = snapshot; return new ConnectorTableSchema("sibling", Collections.emptyList(), "iceberg", Collections.emptyMap(), Collections.emptySet(), "sibling-generation"); } diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.out new file mode 100644 index 00000000000..12364a9a205 --- /dev/null +++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.out @@ -0,0 +1,39 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !iceberg_renamed_nested -- +0 1 x +1 2 y +2 3 z + +-- !iceberg_renamed_nested_topn -- +0 {"a2":1, "b":"x"} +1 {"a2":2, "b":"y"} + +-- !iceberg_update_where_default -- +0 100 42 +1 100 42 +2 2 7 + +-- !iceberg_merge_on_default -- +0 300 42 +1 100 42 +2 2 7 + +-- !gateway_renamed_nested -- +0 1 x +1 2 y +2 3 z + +-- !gateway_renamed_nested_topn -- +0 {"a2":1, "b":"x"} +1 {"a2":2, "b":"y"} + +-- !gateway_update_where_default -- +0 100 42 +1 100 42 +2 2 7 + +-- !gateway_merge_on_default -- +0 300 42 +1 100 42 +2 2 7 + diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.groovy new file mode 100644 index 00000000000..4112cfbb646 --- /dev/null +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_on_hms_gateway_pinned_capabilities.groovy @@ -0,0 +1,118 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// An iceberg table stored in HMS is reachable through a dedicated iceberg catalog ('type'='iceberg', +// 'iceberg.catalog.type'='hms') and through the heterogeneous HMS gateway catalog ('type'='hms'). The gateway +// passes a fixed set of per-table capabilities on from its embedded iceberg sibling to such a table (Top-N lazy +// materialization, nested column pruning, field-id access paths, write-only defaults, ...), and fe-core reads +// them from the schema the statement uses. A statement that reads an iceberg table pins its latest snapshot and +// uses the schema retained with that pin, which the gateway builds through the snapshot overload of +// getTableSchema, as it does for explicit time travel. That overload used to return the sibling's schema without +// the inherited capabilities, so in every such statement an iceberg-on-HMS table planned as if it had none. +// +// The suite builds two identical tables through the dedicated catalog, then drives the same statements through +// the dedicated catalog against one and through the gateway against the other. The plan and error checks are the +// same for both, and the .out file holds the results of both under paired tags, which must be identical. +suite("test_iceberg_on_hms_gateway_pinned_capabilities", "p0,external") { + String enabled = context.config.otherConfigs.get("enableHiveTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable hive test.") + return + } + + String hmsPort = context.config.otherConfigs.get("hive3HmsPort") + String hdfsPort = context.config.otherConfigs.get("hive3HdfsPort") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String icebergCatalog = "test_iceberg_on_hms_pinned_caps_iceberg" + String gatewayCatalog = "test_iceberg_on_hms_pinned_caps_hms" + String dbName = "test_iceberg_on_hms_pinned_caps_db" + + sql """drop catalog if exists ${icebergCatalog}""" + sql """create catalog ${icebergCatalog} properties ( + 'type'='iceberg', + 'iceberg.catalog.type'='hms', + 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}', + 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}' + );""" + sql """drop catalog if exists ${gatewayCatalog}""" + sql """create catalog ${gatewayCatalog} properties ( + 'type'='hms', + 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}', + 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}' + );""" + + // Built through the dedicated catalog for both runs. The first two rows carry d only as the initial default + // 42, and their data files still name the renamed struct field s.a2 as a. Returns the first snapshot. + def createTable = { String tableName -> + sql """switch ${icebergCatalog}""" + sql """create database if not exists ${dbName}""" + sql """use ${dbName}""" + sql """drop table if exists ${tableName}""" + sql """create table ${tableName} (id INT, c INT, s STRUCT) + properties ('format-version' = '3')""" + sql """insert into ${tableName} values (0, 0, named_struct('a', 1, 'b', 'x')), (1, 1, named_struct('a', 2, 'b', 'y'))""" + sql """alter table ${tableName} add column d INT DEFAULT 42""" + sql """alter table ${tableName} rename column s.a TO a2""" + sql """insert into ${tableName} (id, c, s, d) values (2, 2, named_struct('a2', 3, 'b', 'z'), 7)""" + List> snapshots = sql """select snapshot_id from ${tableName}\$snapshots order by committed_at""" + return snapshots[0][0] + } + + def runProbe = { String catalogName, String tableName, String tag -> + def firstSnapshot = createTable(tableName) + sql """switch ${catalogName}""" + sql """use ${dbName}""" + + // Top-N lazy materialization, under the latest-snapshot pin and under explicit time travel. + explain { + sql """select * from ${tableName} order by id limit 1""" + contains "VMaterializeNode" + contains "row_ids: [__DORIS_GLOBAL_ROWID_COL__${tableName}]" + } + explain { + sql """select * from ${tableName} for version as of ${firstSnapshot} order by id limit 1""" + contains "VMaterializeNode" + } + // Nested column pruning. The pruned read of the renamed field needs the field-id access path: by name, + // the old files would not have s.a2. + explain { + sql """select struct_element(s, 'a2') from ${tableName}""" + contains "all access paths: [s.a2]" + } + "order_qt_${tag}_renamed_nested" """select id, struct_element(s, 'a2'), struct_element(s, 'b') from ${tableName}""" + "qt_${tag}_renamed_nested_topn" """select id, s from ${tableName} order by id limit 2""" + + // Write-only defaults: DEFAULT(d) is the write-default 42 in the conditions of UPDATE and MERGE ... + sql """update ${tableName} set c = 100 where d = DEFAULT(d)""" + "order_qt_${tag}_update_where_default" """select id, c, d from ${tableName}""" + sql """merge into ${tableName} t using (select 0 as id) src on t.id = src.id and t.d = DEFAULT(t.d) + when matched then update set c = 300""" + "order_qt_${tag}_merge_on_default" """select id, c, d from ${tableName}""" + // ... and it is refused where the statement writes no row. + test { + sql """delete from ${tableName} where d = DEFAULT(d)""" + exception "DEFAULT(d) cannot be used in a DELETE" + } + test { + sql """select DEFAULT(d) from ${tableName}""" + exception "DEFAULT(d) must reference a column of the table that this statement writes" + } + } + + runProbe(icebergCatalog, "pinned_caps_iceberg", "iceberg") + runProbe(gatewayCatalog, "pinned_caps_gateway", "gateway") +}