diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index e7648c621a6f..1b6f7c725135 100644 --- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -2154,6 +2154,12 @@ public static enum ConfVars { "If this is set to true the URI for auth will have the default location masked with DEFAULT_TABLE_LOCATION"), HIVE_ICEBERG_ALLOW_DATAFILES_IN_TABLE_LOCATION_ONLY("hive.iceberg.allow.datafiles.in.table.location.only", false, "If this is set to true, then all the data files being read should be withing the table location"), + HIVE_ICEBERG_REST_SCAN_PLANNING_MODE( + "hive.iceberg.rest.scan-planning-mode", "client", + new StringSet("client", "server"), + "Hive-wide default for Iceberg REST catalog scan planning in Tez split generation, aligned with\n" + + "catalog property scan-planning-mode. In server mode, scan planning is done by Iceberg REST Server.\n" + + "In client mode (default), split generation is done by Hive."), HIVE_USE_EXPLICIT_RCFILE_HEADER("hive.exec.rcfile.use.explicit.header", true, "If this is set the header for RCFiles will simply be RCF. If this is not\n" + "set the header will be that borrowed from sequence files, e.g. SEQ- followed\n" + diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java index 5d19622f2463..e3aa65754079 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java @@ -284,8 +284,8 @@ public void configureInputJobCredentials(TableDesc tableDesc, MapSerialized tables ({@link SerializableTable}) only carry a metadata snapshot and produce + * {@link org.apache.iceberg.DataTableScan} (client-side manifest planning). When REST catalog + * server-side scan planning is enabled, reload the live table from the catalog so + * {@code table.newScan()} returns {@link org.apache.iceberg.rest.RESTTableScan} and issues + * {@code POST /plan} on the REST server. + * + *

Intra-transaction read-after-write ({@link InputFormatConfig#TABLE_METADATA_LOCATION}) still + * uses the deserialized snapshot so uncommitted metadata is visible. + */ + public static Table resolveTableForScanPlanning(Configuration conf, String tableIdentifier) { + if (shouldReloadForServerSideScanPlanning(conf)) { + Table table = Catalogs.loadTable(conf); + checkAndSetIoConfig(conf, table); + IcebergVendedCredentialUtil.applyFromJobConf(table, resolveCatalogName(conf), conf); + return table; + } + + Table table = deserializeTable(conf, tableIdentifier); + if (table == null) { + table = Catalogs.loadTable(conf); + checkAndSetIoConfig(conf, table); + } + return table; + } + + private static boolean shouldReloadForServerSideScanPlanning(Configuration conf) { + if (StringUtils.isNotBlank(conf.get(InputFormatConfig.TABLE_METADATA_LOCATION))) { + return false; + } + String catalogName = resolveCatalogName(conf); + if (StringUtils.isBlank(catalogName)) { + return false; + } + if (!CatalogUtil.ICEBERG_CATALOG_TYPE_REST.equals( + IcebergCatalogProperties.getCatalogType(conf, catalogName))) { + return false; + } + return RestCatalogScanPlanningUtil.isServerSidePlanningEnabled(catalogName, conf); + } + + /** + * Resolves the Iceberg catalog name from {@link InputFormatConfig#CATALOG_NAME} on the + * configuration, or the session default ({@link IcebergCatalogProperties#getCatalogName(Configuration)}). + */ + public static String resolveCatalogName(Configuration conf) { + return resolveCatalogName(conf, conf != null ? conf.get(InputFormatConfig.CATALOG_NAME) : null); + } + + /** + * Resolves the Iceberg catalog name from a per-table {@link InputFormatConfig#CATALOG_NAME} value + * (e.g. {@code TableDesc} properties) or the session default when that value is absent. + */ + public static String resolveCatalogName(Configuration conf, String catalogNameFromTable) { + if (StringUtils.isNotBlank(catalogNameFromTable)) { + return catalogNameFromTable; + } + if (conf == null) { + return null; + } + return IcebergCatalogProperties.getCatalogName(conf); + } + /** * If enabled, it populates the FileIO's hadoop configuration with the input config object. * This might be necessary when the table object was serialized without the FileIO config. diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/RestCatalogScanPlanningUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/RestCatalogScanPlanningUtil.java new file mode 100644 index 000000000000..347ae8a186f6 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/RestCatalogScanPlanningUtil.java @@ -0,0 +1,206 @@ +/* + * 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. + */ + +package org.apache.iceberg.mr.hive; + +import java.util.Map; +import java.util.function.BiConsumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.rest.RESTCatalogProperties; + +/** + * Utilities for Iceberg REST catalog server-side scan planning configuration. + * + *

When a REST catalog server advertises the scan-planning endpoints and + * {@link RESTCatalogProperties#SCAN_PLANNING_MODE} is set to + * {@link RESTCatalogProperties.ScanPlanningMode#SERVER}, Iceberg's {@code RESTSessionCatalog} returns a + * {@code RESTTable} that delegates {@code planTasks()} to the server. Hive's + * {@code IcebergInputFormat} calls {@code scan.planTasks()} via + * {@link HiveTableUtil#resolveTableForScanPlanning}, which reloads the live REST catalog table + * (instead of a serialized metadata snapshot) when server mode is enabled and + * {@link HiveConf.ConfVars#HIVE_ICEBERG_REST_SCAN_PLANNING_MODE} is {@code server}. + * Operators can use this helper or set catalog {@code scan-planning-mode} directly in {@code hive-site.xml}. + * + *

Tests: {@code TestRestCatalogScanPlanningUtil} in {@code iceberg-handler}; + * {@code TestHiveIcebergServerSideScanPlanning} in {@code iceberg-handler}; embedded REST server + * tests {@code TestRestCatalogScanPlanningServerIT} and {@code TestHiveIcebergServerSideScanPlanningServerIT} + * in {@code itests/hive-iceberg-rest-server}. + * + *

{@link #setCatalogMode(Configuration, String, String)}, + * {@link #isCatalogServerMode(Configuration, String)}, and + * {@link #setHiveMode(Configuration, String)} are public so tests in other Maven modules + * (for example {@code itests/hive-iceberg-rest-server}) can configure scan planning; those modules + * compile against this artifact as a JAR and cannot call package-private members. They are not + * intended as a general operator or application API. + * + * @see REST catalog properties + */ +public final class RestCatalogScanPlanningUtil { + + private RestCatalogScanPlanningUtil() { + } + + private static void setCatalogMode( + Configuration conf, String catalogName, RESTCatalogProperties.ScanPlanningMode mode) { + conf.set( + IcebergCatalogProperties.catalogPropertyConfigKey( + catalogName, RESTCatalogProperties.SCAN_PLANNING_MODE), + mode.modeName()); + } + + public static void setCatalogMode(Configuration conf, String catalogName, String mode) { + setCatalogMode(conf, catalogName, RESTCatalogProperties.ScanPlanningMode.fromString(mode)); + } + + static RESTCatalogProperties.ScanPlanningMode getCatalogMode( + Configuration conf, String catalogName) { + String mode = conf.get( + IcebergCatalogProperties.catalogPropertyConfigKey( + catalogName, RESTCatalogProperties.SCAN_PLANNING_MODE), + RESTCatalogProperties.SCAN_PLANNING_MODE_DEFAULT.modeName()); + return RESTCatalogProperties.ScanPlanningMode.fromString(mode); + } + + public static boolean isCatalogServerMode(Configuration conf, String catalogName) { + return getCatalogMode(conf, catalogName) == RESTCatalogProperties.ScanPlanningMode.SERVER; + } + + /** + * Returns true when Hive server-side REST scan planning is enabled in configuration. + */ + static boolean isHiveServerMode(Configuration conf) { + if (conf == null) { + return false; + } + return RESTCatalogProperties.ScanPlanningMode.SERVER == + RESTCatalogProperties.ScanPlanningMode.fromString(getHiveMode(conf)); + } + + static String getHiveMode(Configuration conf) { + if (conf == null) { + return RESTCatalogProperties.SCAN_PLANNING_MODE_DEFAULT.modeName(); + } + return HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SCAN_PLANNING_MODE); + } + + public static void setHiveMode(Configuration conf, String mode) { + HiveConf.setVar( + conf, + HiveConf.ConfVars.HIVE_ICEBERG_REST_SCAN_PLANNING_MODE, + RESTCatalogProperties.ScanPlanningMode.fromString(mode).modeName()); + } + + /** + * Returns true when the catalog is configured for server-side scan planning and the Hive feature flag is on. + */ + static boolean isServerSidePlanningEnabled(String catalogName, Configuration conf) { + if (conf == null || StringUtils.isEmpty(catalogName)) { + return false; + } + return isHiveServerMode(conf) && isCatalogServerMode(conf, catalogName); + } + + /** + * Returns true when catalog properties should be copied into the Tez job configuration so + * executors can reload a live REST catalog table for server-side scan planning. + */ + static boolean shouldPropagateCatalogPropertiesToJob(String catalogName, Configuration conf) { + String resolvedCatalogName = HiveTableUtil.resolveCatalogName(conf, catalogName); + if (StringUtils.isEmpty(resolvedCatalogName) || conf == null) { + return false; + } + if (!CatalogUtil.ICEBERG_CATALOG_TYPE_REST.equals( + IcebergCatalogProperties.getCatalogType(conf, resolvedCatalogName))) { + return false; + } + return isServerSidePlanningEnabled(resolvedCatalogName, conf); + } + + /** + * Copies {@code iceberg.catalog..*} entries from the HS2 session configuration into Tez + * job properties so executors can reload a live REST catalog table for server-side scan planning. + * + *

Session-level {@code SET} commands and {@code hive-site.xml} catalog settings are not + * automatically present in the job configuration; without this step split generation falls back to + * the serialized metadata snapshot ({@code DataTableScan}). + */ + public static void propagateCatalogPropertiesToJob( + Configuration sessionConf, String catalogName, Map jobProperties) { + if (sessionConf == null || jobProperties == null) { + return; + } + propagateCatalogPropertiesToJob(sessionConf, catalogName, jobProperties::putIfAbsent); + } + + /** + * Copies REST catalog configuration from the HS2 session into a runtime job {@link Configuration}. + * Called from {@code configureJobConf} so split generation sees catalog URI/type/scan-planning-mode + * even when job properties were not copied yet. + */ + public static void propagateCatalogPropertiesToJob( + Configuration sessionConf, String catalogName, Configuration jobConf) { + if (sessionConf == null || jobConf == null) { + return; + } + propagateCatalogPropertiesToJob( + sessionConf, + catalogName, + (key, value) -> { + if (jobConf.get(key) == null) { + jobConf.set(key, value); + } + }); + } + + private static void propagateCatalogPropertiesToJob( + Configuration sessionConf, String catalogName, BiConsumer jobPropertySink) { + if (!shouldPropagateCatalogPropertiesToJob(catalogName, sessionConf)) { + return; + } + + String resolvedCatalogName = HiveTableUtil.resolveCatalogName(sessionConf, catalogName); + if (StringUtils.isEmpty(resolvedCatalogName)) { + return; + } + + jobPropertySink.accept( + HiveConf.ConfVars.HIVE_ICEBERG_REST_SCAN_PLANNING_MODE.varname, + getHiveMode(sessionConf)); + + String sessionDefaultCatalog = + MetastoreConf.getVar(sessionConf, MetastoreConf.ConfVars.CATALOG_DEFAULT); + if (StringUtils.isNotBlank(sessionDefaultCatalog)) { + jobPropertySink.accept(MetastoreConf.ConfVars.CATALOG_DEFAULT.getVarname(), sessionDefaultCatalog); + } + + String catalogPrefix = + IcebergCatalogProperties.CATALOG_CONFIG_PREFIX + resolvedCatalogName + "."; + sessionConf.forEach( + entry -> { + if (entry.getKey().startsWith(catalogPrefix)) { + jobPropertySink.accept(entry.getKey(), entry.getValue()); + } + }); + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java index 89f888255b89..f2a388dd6418 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java @@ -22,7 +22,6 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.List; -import java.util.Optional; import java.util.concurrent.ExecutorService; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; @@ -56,7 +55,6 @@ import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.io.CloseableIterable; -import org.apache.iceberg.mr.Catalogs; import org.apache.iceberg.mr.InputFormatConfig; import org.apache.iceberg.mr.hive.HiveTableUtil; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -157,16 +155,18 @@ private static > T applyConfig @Override public List getSplits(JobContext context) { Configuration conf = context.getConfiguration(); - Table table = Optional - .ofNullable(HiveTableUtil.deserializeTable(conf, conf.get(InputFormatConfig.TABLE_IDENTIFIER))) - .orElseGet(() -> { - Table tbl = Catalogs.loadTable(conf); - conf.set(InputFormatConfig.TABLE_IDENTIFIER, tbl.name()); - // planning-local conf only (never shipped): for credential-vending catalogs the loaded - // table's FileIO carries secrets, which must not reach a serialized job configuration - conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tbl.name(), SerializationUtil.serializeToBase64(tbl)); - return tbl; - }); + String tableIdentifier = conf.get(InputFormatConfig.TABLE_IDENTIFIER); + Table table = HiveTableUtil.resolveTableForScanPlanning(conf, tableIdentifier); + if (StringUtils.isBlank(tableIdentifier)) { + tableIdentifier = table.name(); + conf.set(InputFormatConfig.TABLE_IDENTIFIER, tableIdentifier); + } + String serializedTableKey = InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableIdentifier; + if (conf.get(serializedTableKey) == null) { + // planning-local conf only (never shipped): for credential-vending catalogs the loaded + // table's FileIO carries secrets, which must not reach a serialized job configuration + conf.set(serializedTableKey, SerializationUtil.serializeToBase64(table)); + } final ExecutorService workerPool = ThreadPools.newFixedThreadPool("iceberg-plan-worker-pool", conf.getInt(SystemConfigs.WORKER_THREAD_POOL_SIZE.propertyKey(), ThreadPools.WORKER_THREAD_POOL_SIZE)); diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestRestCatalogScanPlanningUtil.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestRestCatalogScanPlanningUtil.java new file mode 100644 index 000000000000..d43eb59f4733 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestRestCatalogScanPlanningUtil.java @@ -0,0 +1,204 @@ +/* + * 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. + */ + +package org.apache.iceberg.mr.hive; + +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.rest.RESTCatalogProperties; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link RestCatalogScanPlanningUtil}: Hive configuration keys, server-mode detection, + * and copying REST catalog settings from the HS2 session into Tez/MR job configuration. End-to-end + * {@code POST /plan} behavior is in {@code TestRestCatalogScanPlanningServerIT} + * ({@code itests/hive-iceberg-rest-server}). + */ +public class TestRestCatalogScanPlanningUtil { + + /** + * {@link RestCatalogScanPlanningUtil#setCatalogMode} must use the standard + * {@code iceberg.catalog..*} prefix so {@code scan-planning-mode} is read/written consistently + * with other Iceberg catalog properties in {@code hive-site.xml} and session {@code SET}. + */ + @Test + void setCatalogModeUsesCatalogPropertyConfigKey() { + Configuration conf = new Configuration(); + RestCatalogScanPlanningUtil.setCatalogMode(conf, "ice01", "server"); + assertThat(conf.get( + IcebergCatalogProperties.catalogPropertyConfigKey( + "ice01", RESTCatalogProperties.SCAN_PLANNING_MODE))) + .isEqualTo("server"); + } + + /** + * {@code scan-planning-mode=server} alone is not enough; {@link HiveConf.ConfVars + * #HIVE_ICEBERG_REST_SCAN_PLANNING_MODE} must also be {@code server}. + */ + @Test + void isServerSidePlanningEnabledFromConfiguration() { + Configuration conf = new Configuration(); + assertThat(RestCatalogScanPlanningUtil.isServerSidePlanningEnabled("ice01", conf)).isFalse(); + + RestCatalogScanPlanningUtil.setCatalogMode(conf, "ice01", "server"); + assertThat(RestCatalogScanPlanningUtil.isServerSidePlanningEnabled("ice01", conf)).isFalse(); + assertThat(RestCatalogScanPlanningUtil.isCatalogServerMode(conf, "ice01")).isTrue(); + + enableHiveServerMode(conf); + assertThat(RestCatalogScanPlanningUtil.isServerSidePlanningEnabled("ice01", conf)).isTrue(); + assertThat(RestCatalogScanPlanningUtil.getCatalogMode(conf, "ice01").modeName()) + .isEqualTo("server"); + } + + /** + * Job propagation uses the per-table catalog name when present; otherwise the session default + * catalog from {@link MetastoreConf.ConfVars#CATALOG_DEFAULT}. + */ + @Test + void resolveCatalogNameUsesSessionDefaultWhenTablePropertyMissing() { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, "ice01"); + assertThat(HiveTableUtil.resolveCatalogName(conf, null)).isEqualTo("ice01"); + assertThat(HiveTableUtil.resolveCatalogName(conf, "ice02")).isEqualTo("ice02"); + } + + /** + * Catalog properties are copied to the job only for REST catalogs with server scan planning; + * Hive (metastore) catalogs and REST catalogs in local planning mode are skipped. + */ + @Test + void shouldPropagateCatalogPropertiesOnlyForRestCatalogInServerMode() { + Configuration conf = new Configuration(); + conf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + assertThat(RestCatalogScanPlanningUtil.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isFalse(); + + RestCatalogScanPlanningUtil.setCatalogMode(conf, "ice01", "server"); + assertThat(RestCatalogScanPlanningUtil.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isFalse(); + enableHiveServerMode(conf); + assertThat(RestCatalogScanPlanningUtil.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isTrue(); + + Configuration hiveConf = new Configuration(); + hiveConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE); + RestCatalogScanPlanningUtil.setCatalogMode(hiveConf, "ice01", "server"); + enableHiveServerMode(hiveConf); + assertThat(RestCatalogScanPlanningUtil.shouldPropagateCatalogPropertiesToJob("ice01", hiveConf)) + .isFalse(); + } + + /** + * When propagation is enabled, all {@code iceberg.catalog..*} entries from the HS2 session + * (type, URI, scan-planning mode, etc.) are copied into Tez/MR job properties; unrelated conf keys + * are not. + */ + @Test + void propagateCatalogPropertiesToJobCopiesRestCatalogSettings() { + Configuration sessionConf = new Configuration(); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", "uri"), "http://localhost:8181"); + RestCatalogScanPlanningUtil.setCatalogMode(sessionConf, "ice01", "server"); + enableHiveServerMode(sessionConf); + sessionConf.set("unrelated.key", "skip"); + + Map jobProperties = Maps.newHashMap(); + RestCatalogScanPlanningUtil.propagateCatalogPropertiesToJob(sessionConf, "ice01", jobProperties); + + assertThat(jobProperties) + .containsEntry( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST) + .containsEntry( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", "uri"), "http://localhost:8181") + .containsEntry( + IcebergCatalogProperties.catalogPropertyConfigKey( + "ice01", RESTCatalogProperties.SCAN_PLANNING_MODE), + "server") + .containsEntry( + HiveConf.ConfVars.HIVE_ICEBERG_REST_SCAN_PLANNING_MODE.varname, "server") + .doesNotContainKey("unrelated.key"); + } + + /** + * Same as {@link #propagateCatalogPropertiesToJobCopiesRestCatalogSettings()} for the + * {@link Configuration} overload used by {@code HiveIcebergStorageHandler#configureJobConf}. + */ + @Test + void propagateCatalogPropertiesToJobConfigurationCopiesRestCatalogSettings() { + Configuration sessionConf = new Configuration(); + MetastoreConf.setVar(sessionConf, MetastoreConf.ConfVars.CATALOG_DEFAULT, "ice01"); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", "uri"), "http://localhost:8181"); + RestCatalogScanPlanningUtil.setCatalogMode(sessionConf, "ice01", "server"); + enableHiveServerMode(sessionConf); + + Configuration jobConf = new Configuration(); + RestCatalogScanPlanningUtil.propagateCatalogPropertiesToJob(sessionConf, null, jobConf); + + assertThat(jobConf.get(MetastoreConf.ConfVars.CATALOG_DEFAULT.getVarname())).isEqualTo("ice01"); + assertThat(jobConf.get( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE))) + .isEqualTo(CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + assertThat(jobConf.get( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", "uri"))) + .isEqualTo("http://localhost:8181"); + assertThat(jobConf.get( + IcebergCatalogProperties.catalogPropertyConfigKey( + "ice01", RESTCatalogProperties.SCAN_PLANNING_MODE))) + .isEqualTo("server"); + } + + /** + * Negative path: REST catalog settings are not copied when {@code scan-planning-mode} is not + * {@code server}, so executors keep using the serialized table snapshot for split generation. + */ + @Test + void propagateCatalogPropertiesToJobSkipsWhenServerModeDisabled() { + Configuration sessionConf = new Configuration(); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", "uri"), "http://localhost:8181"); + + Map jobProperties = Maps.newHashMap(); + RestCatalogScanPlanningUtil.propagateCatalogPropertiesToJob(sessionConf, "ice01", jobProperties); + + assertThat(jobProperties).isEmpty(); + } + + private static void enableHiveServerMode(Configuration conf) { + RestCatalogScanPlanningUtil.setHiveMode(conf, "server"); + } +} diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanning.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanning.java new file mode 100644 index 000000000000..d3fa81cd555a --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanning.java @@ -0,0 +1,104 @@ +/* + * 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. + */ + +package org.apache.iceberg.rest; + +import java.nio.file.Path; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SerializableTable; +import org.apache.iceberg.Table; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.hive.HiveTableUtil; +import org.apache.iceberg.mr.hive.RestCatalogScanPlanningUtil; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for reload guards on {@link HiveTableUtil#resolveTableForScanPlanning} (see + * method-level comments). End-to-end Hive server-side scan planning is documented on + * {@code TestHiveIcebergServerSideScanPlanningServerIT}. + */ +class TestHiveIcebergServerSideScanPlanning { + + private static final String CATALOG_NAME = "ice01"; + private static final Schema SCHEMA = + new Schema(Types.NestedField.required(1, "id", Types.LongType.get())); + + @TempDir + private Path warehouse; + + /** + * Negative path: do not reload from the REST catalog unless {@code scan-planning-mode=server}. + * Even with a REST catalog and a serialized table in the job conf, split generation should keep + * using the snapshot ({@link SerializableTable}) when server mode is off. + */ + @Test + void usesDeserializedTableWhenServerModeDisabled() { + Table table = new HadoopTables().create(SCHEMA, PartitionSpec.unpartitioned(), warehouse.toString()); + Configuration conf = buildTableConf(table, false); + + Table resolved = HiveTableUtil.resolveTableForScanPlanning(conf, table.name()); + assertThat(resolved).isInstanceOf(SerializableTable.class); + } + + /** + * Negative path: intra-transaction read-after-write must not reload from the catalog even when + * server mode is enabled. {@link InputFormatConfig#TABLE_METADATA_LOCATION} points at uncommitted + * metadata that only exists in the job conf; reloading from the REST catalog would return stale + * committed state and break same-txn visibility (e.g. INSERT then SELECT). + */ + @Test + void usesDeserializedTableForIntraTxnMetadataEvenInServerMode() { + Table table = new HadoopTables().create(SCHEMA, PartitionSpec.unpartitioned(), warehouse.toString()); + Configuration conf = buildTableConf(table, true); + conf.set(InputFormatConfig.TABLE_METADATA_LOCATION, warehouse + "/metadata/snapshot.metadata.json"); + + Table resolved = HiveTableUtil.resolveTableForScanPlanning(conf, table.name()); + assertThat(resolved).isInstanceOf(BaseTable.class); + } + + private Configuration buildTableConf(Table table, boolean serverMode) { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + conf.set(InputFormatConfig.TABLE_IDENTIFIER, table.name()); + conf.set(InputFormatConfig.TABLE_LOCATION, warehouse.toString()); + conf.set(InputFormatConfig.CATALOG_NAME, CATALOG_NAME); + conf.set( + IcebergCatalogProperties.catalogPropertyConfigKey(CATALOG_NAME, CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + if (serverMode) { + RestCatalogScanPlanningUtil.setCatalogMode(conf, CATALOG_NAME, "server"); + RestCatalogScanPlanningUtil.setHiveMode(conf, "server"); + } + conf.set( + InputFormatConfig.SERIALIZED_TABLE_PREFIX + table.name(), + HiveTableUtil.serializeTable(table, conf, null, null)); + return conf; + } +} diff --git a/iceberg/iceberg-rest-catalog-client/pom.xml b/iceberg/iceberg-rest-catalog-client/pom.xml index a41437f881f2..cb73dbabb3ed 100644 --- a/iceberg/iceberg-rest-catalog-client/pom.xml +++ b/iceberg/iceberg-rest-catalog-client/pom.xml @@ -75,17 +75,5 @@ tests test - - org.apache.iceberg - iceberg-core - tests - test - - - org.apache.iceberg - iceberg-api - - - diff --git a/itests/hive-iceberg-rest-server/pom.xml b/itests/hive-iceberg-rest-server/pom.xml new file mode 100644 index 000000000000..5c49a4676e42 --- /dev/null +++ b/itests/hive-iceberg-rest-server/pom.xml @@ -0,0 +1,234 @@ + + + + 4.0.0 + + org.apache.hive + hive-it + 4.3.0-SNAPSHOT + ../pom.xml + + hive-it-iceberg-rest-server + jar + Hive Iceberg Integration - In-process REST catalog server tests + + Integration tests that use Iceberg's in-process REST catalog server (for example + TestBaseWithRESTServer and the iceberg-core REST test harness) together with Hive + Iceberg executor code. Test classpath uses unshaded Iceberg and unpacks only + org.apache.iceberg.mr classes from hive-iceberg-handler so the REST harness is not + mixed with the shaded handler jar. For Hive metastore REST catalog client tests + (Keycloak, Testcontainers, and similar), see hive-it-iceberg. + + + ../.. + UTF-8 + + 12.1.8 + + + + + org.eclipse.jetty + jetty-http + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty + jetty-io + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty + jetty-util + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty + jetty-server + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty + jetty-security + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty.ee10 + jetty-ee10-servlet + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty.compression + jetty-compression-server + ${iceberg.rest.test.jetty.version} + + + org.eclipse.jetty.compression + jetty-compression-gzip + ${iceberg.rest.test.jetty.version} + + + + + + org.apache.hive + hive-iceberg-rest-catalog-client + ${project.version} + + + org.apache.hive + hive-iceberg-shading + + + + + org.apache.hive + hive-iceberg-handler + ${project.version} + test + + + org.apache.hive + patched-iceberg-core + test + + + org.apache.hive + patched-iceberg-api + test + + + org.apache.iceberg + iceberg-core + ${iceberg.version} + tests + test + + + org.junit.jupiter + junit-jupiter + ${junit.jupiter.version} + test + + + org.mockito + mockito-core + test + + + org.assertj + assertj-core + test + + + jakarta.servlet + jakarta.servlet-api + test + + + org.eclipse.jetty + jetty-http + test + + + org.eclipse.jetty + jetty-io + test + + + org.eclipse.jetty + jetty-util + test + + + org.eclipse.jetty + jetty-server + test + + + org.eclipse.jetty.ee10 + jetty-ee10-servlet + test + + + + org.eclipse.jetty + jetty-security + + + + + org.eclipse.jetty + jetty-security + test + + + org.eclipse.jetty.compression + jetty-compression-server + test + + + org.eclipse.jetty.compression + jetty-compression-gzip + test + + + + + + org.apache.maven.plugins + maven-dependency-plugin + + + unpack-handler-mr-classes + generate-test-resources + + unpack + + + + + org.apache.hive + hive-iceberg-handler + ${project.version} + org/apache/iceberg/mr/** + ${project.build.directory}/handler-mr-classes + + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + + org.apache.hive:hive-iceberg-shading + org.apache.hive:hive-iceberg-handler + + + ${project.build.directory}/handler-mr-classes + + + + + + diff --git a/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanningServerIT.java b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanningServerIT.java new file mode 100644 index 000000000000..cb10b2e1829c --- /dev/null +++ b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanningServerIT.java @@ -0,0 +1,219 @@ +/* + * 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. + */ + +package org.apache.iceberg.rest; + +import java.io.IOException; +import java.util.Map; +import java.util.function.Consumer; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Scan; +import org.apache.iceberg.SerializableTable; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.mr.hive.RestCatalogScanPlanningUtil; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.hive.HiveTableUtil; +import org.apache.iceberg.rest.responses.ErrorResponse; +import org.apache.iceberg.rest.responses.LoadTableResponse; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import static org.apache.iceberg.TestBase.FILE_A; +import static org.apache.iceberg.TestBase.SCHEMA; +import static org.apache.iceberg.TestBase.SPEC; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.verify; + +/** + * Integration tests for Hive server-side REST catalog scan planning via + * {@link HiveTableUtil#resolveTableForScanPlanning}: with catalog settings propagated from HS2 into + * the executor job conf, split generation reloads a live {@link RESTTable} from the embedded REST + * server instead of the serialized metadata snapshot ({@link SerializableTable}). + * + *

Without propagation, or when reload guards apply, executors keep the snapshot; those cases are + * unit-tested in {@code TestHiveIcebergServerSideScanPlanning} ({@code iceberg-handler}). Iceberg + * REST client {@code POST /plan} behavior is in {@code TestRestCatalogScanPlanningServerIT}. + */ +class TestHiveIcebergServerSideScanPlanningServerIT extends TestBaseWithRESTServer { + + private static final String CATALOG_NAME = "hive-iceberg-scan-planning"; + private static final TableIdentifier TABLE_ID = TableIdentifier.of(NS, "file_planning_table"); + + @Override + protected boolean useHttpCompression() { + return false; + } + + @Override + protected RESTCatalogAdapter createAdapterForServer() { + return Mockito.spy( + new RESTCatalogAdapter(backendCatalog) { + @Override + protected T execute( + HTTPRequest request, + Class responseType, + Consumer errorHandler, + Consumer> responseHeaders) { + Object body = roundTripSerialize(request.body(), "request"); + HTTPRequest req = ImmutableHTTPRequest.builder().from(request).body(body).build(); + T response = super.execute(req, responseType, errorHandler, responseHeaders); + response = roundTripSerialize(response, "response"); + + if (response instanceof LoadTableResponse) { + return RESTCatalogAdapter.castResponse( + responseType, + withPlanningMode( + (LoadTableResponse) response, + RESTCatalogProperties.ScanPlanningMode.SERVER.modeName())); + } + + return response; + } + }); + } + + @Override + protected String catalogName() { + return CATALOG_NAME; + } + + @Override + protected Map additionalCatalogProperties() { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + RestCatalogScanPlanningUtil.setCatalogMode(conf, CATALOG_NAME, "server"); + RestCatalogScanPlanningUtil.setHiveMode(conf, "server"); + return IcebergCatalogProperties.getCatalogProperties(conf, CATALOG_NAME); + } + + @BeforeEach + @Override + public void before() throws Exception { + super.before(); + adapterForRESTServer.setPlanningBehavior( + new RESTCatalogAdapter.PlanningBehavior() { + @Override + public boolean shouldPlanTableScanAsync(Scan scan) { + return false; + } + + @Override + public int numberFileScanTasksPerPlanTask() { + return 100; + } + }); + } + + /** + * Negative path: executor job conf without propagated catalog settings must not reload from the + * REST catalog, even when server mode is enabled on HS2. Without catalog URI/type in the job + * conf, split generation falls back to the serialized table snapshot. + */ + @Test + void executorJobConfWithoutPropagationUsesSerializedTable() throws IOException { + Table table = createTableWithData(); + Configuration jobConf = executorJobConf(table); + + Table resolved = HiveTableUtil.resolveTableForScanPlanning(jobConf, TABLE_ID.toString()); + assertThat(resolved).isInstanceOf(SerializableTable.class); + assertThat(resolved.newScan()).isNotInstanceOf(RESTTableScan.class); + } + + /** + * Positive path: when catalog properties are propagated from HS2 into the executor job conf, + * split generation reloads a live {@link RESTTable} from the REST catalog instead of using the + * serialized metadata snapshot. + */ + @Test + void propagatedJobConfReloadsRestTableFromCatalog() throws IOException { + Table table = createTableWithData(); + Configuration sessionConf = sessionConf(); + Configuration jobConf = executorJobConf(table); + RestCatalogScanPlanningUtil.propagateCatalogPropertiesToJob(sessionConf, CATALOG_NAME, jobConf); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HTTPRequest.class); + Table resolved = HiveTableUtil.resolveTableForScanPlanning(jobConf, TABLE_ID.toString()); + + assertThat(resolved).isInstanceOf(RESTTable.class); + assertThat(resolved.newScan()).isInstanceOf(RESTTableScan.class); + + verify(adapterForRESTServer, atLeastOnce()) + .execute(requestCaptor.capture(), any(), any(), any(), any()); + assertThat( + requestCaptor.getAllValues().stream() + .anyMatch(req -> req.path().contains("/tables/"))) + .as("Expected Hive split planning to reload the table from the REST catalog") + .isTrue(); + } + + private Table createTableWithData() { + restCatalog.createNamespace(NS); + Table table = + restCatalog.buildTable(TABLE_ID, SCHEMA).withPartitionSpec(SPEC).create(); + table.newAppend().appendFile(FILE_A).commit(); + return table; + } + + private Configuration sessionConf() { + Configuration sessionConf = new Configuration(); + MetastoreConf.setVar(sessionConf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey(CATALOG_NAME, CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_REST); + restCatalog.properties().forEach( + (key, value) -> + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey(CATALOG_NAME, key), value)); + sessionConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey(CATALOG_NAME, CatalogProperties.URI), + httpServer.getURI().toString()); + RestCatalogScanPlanningUtil.setCatalogMode(sessionConf, CATALOG_NAME, "server"); + RestCatalogScanPlanningUtil.setHiveMode(sessionConf, "server"); + return sessionConf; + } + + private Configuration executorJobConf(Table table) throws IOException { + Configuration jobConf = new Configuration(); + jobConf.set(InputFormatConfig.TABLE_IDENTIFIER, TABLE_ID.toString()); + jobConf.set(InputFormatConfig.CATALOG_NAME, CATALOG_NAME); + jobConf.set( + InputFormatConfig.SERIALIZED_TABLE_PREFIX + TABLE_ID.toString(), + HiveTableUtil.serializeTable(table, jobConf, null, null)); + return jobConf; + } + + private static LoadTableResponse withPlanningMode(LoadTableResponse response, String mode) { + return LoadTableResponse.builder() + .withTableMetadata(response.tableMetadata()) + .addAllConfig(response.config()) + .addConfig(RESTCatalogProperties.SCAN_PLANNING_MODE, mode) + .addAllCredentials(response.credentials()) + .build(); + } +} diff --git a/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanningServerIT.java b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanningServerIT.java new file mode 100644 index 000000000000..10d1e00ce969 --- /dev/null +++ b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanningServerIT.java @@ -0,0 +1,170 @@ +/* + * 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. + */ + +package org.apache.iceberg.rest; + +import java.io.IOException; +import java.util.Map; +import java.util.function.Consumer; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Scan; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.mr.hive.RestCatalogScanPlanningUtil; +import org.apache.iceberg.rest.requests.PlanTableScanRequest; +import org.apache.iceberg.rest.responses.ErrorResponse; +import org.apache.iceberg.rest.responses.LoadTableResponse; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import static org.apache.iceberg.TestBase.FILE_A; +import static org.apache.iceberg.TestBase.SCHEMA; +import static org.apache.iceberg.TestBase.SPEC; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.verify; + +/** + * Embedded REST server tests for {@link RestCatalogScanPlanningUtil}: when Hive-style configuration sets + * {@code scan-planning-mode=server}, {@link RESTTable} / {@link RESTTableScan} delegate split + * planning to the catalog server ({@code POST /plan}). + */ +class TestRestCatalogScanPlanningServerIT extends TestBaseWithRESTServer { + + private static final String CATALOG_NAME = "hive-rest-scan-planning"; + + @Override + protected RESTCatalogAdapter createAdapterForServer() { + return Mockito.spy( + new RESTCatalogAdapter(backendCatalog) { + @Override + protected T execute( + HTTPRequest request, + Class responseType, + Consumer errorHandler, + Consumer> responseHeaders) { + Object body = roundTripSerialize(request.body(), "request"); + HTTPRequest req = ImmutableHTTPRequest.builder().from(request).body(body).build(); + T response = super.execute(req, responseType, errorHandler, responseHeaders); + + if (response instanceof LoadTableResponse) { + return RESTCatalogAdapter.castResponse( + responseType, + withPlanningMode( + (LoadTableResponse) response, + RESTCatalogProperties.ScanPlanningMode.SERVER.modeName())); + } + + if (req.body() instanceof PlanTableScanRequest) { + return response; + } + + return roundTripSerialize(response, "response"); + } + }); + } + + @Override + protected String catalogName() { + return CATALOG_NAME; + } + + @Override + protected Map additionalCatalogProperties() { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + RestCatalogScanPlanningUtil.setCatalogMode(conf, CATALOG_NAME, "server"); + return IcebergCatalogProperties.getCatalogProperties(conf, CATALOG_NAME); + } + + @BeforeEach + @Override + public void before() throws Exception { + super.before(); + adapterForRESTServer.setPlanningBehavior( + new RESTCatalogAdapter.PlanningBehavior() { + @Override + public boolean shouldPlanTableScanAsync(Scan scan) { + return false; + } + + @Override + public int numberFileScanTasksPerPlanTask() { + return 100; + } + }); + } + + /** + * Positive path: REST catalog loaded with server planning mode returns a {@link RESTTable}; calling + * {@code planTasks()} on a scan issues a {@link PlanTableScanRequest} to the embedded REST server. + */ + @Test + void hiveCatalogConfigurationIssuesPlanTableScanRequest() { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + RestCatalogScanPlanningUtil.setCatalogMode(conf, CATALOG_NAME, "server"); + RestCatalogScanPlanningUtil.setHiveMode(conf, "server"); + assertThat(IcebergCatalogProperties.getCatalogProperties(conf, CATALOG_NAME)) + .containsEntry(RESTCatalogProperties.SCAN_PLANNING_MODE, "server"); + assertThat(RestCatalogScanPlanningUtil.isCatalogServerMode(conf, CATALOG_NAME)).isTrue(); + + restCatalog.createNamespace(NS); + Table table = + restCatalog.buildTable(TableIdentifier.of(NS, "scan_planning_table"), SCHEMA) + .withPartitionSpec(SPEC) + .create(); + table.newAppend().appendFile(FILE_A).commit(); + + parserContext = + ParserContext.builder() + .add("specsById", table.specs()) + .add("caseSensitive", false) + .build(); + + assertThat(table).isInstanceOf(RESTTable.class); + assertThat(table.newScan()).isInstanceOf(RESTTableScan.class); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HTTPRequest.class); + assertThat(table.newScan().planTasks()).isNotEmpty(); + + verify(adapterForRESTServer, atLeastOnce()) + .execute(requestCaptor.capture(), any(), any(), any(), any()); + assertThat( + requestCaptor.getAllValues().stream() + .anyMatch(req -> req.body() instanceof PlanTableScanRequest)) + .as("Expected server-side scan planning via POST /plan (PlanTableScanRequest)") + .isTrue(); + } + + private static LoadTableResponse withPlanningMode(LoadTableResponse response, String mode) { + return LoadTableResponse.builder() + .withTableMetadata(response.tableMetadata()) + .addAllConfig(response.config()) + .addConfig(RESTCatalogProperties.SCAN_PLANNING_MODE, mode) + .addAllCredentials(response.credentials()) + .build(); + } +} diff --git a/itests/pom.xml b/itests/pom.xml index 6c64b0d730c3..3d6875511234 100644 --- a/itests/pom.xml +++ b/itests/pom.xml @@ -50,6 +50,7 @@ qtest-iceberg test-docker hive-iceberg + hive-iceberg-rest-server @@ -491,12 +492,32 @@ hive-iceberg-handler ${project.version} + + org.apache.hive + patched-iceberg-api + ${project.version} + + + org.apache.hive + patched-iceberg-core + ${project.version} + + + jakarta.servlet + jakarta.servlet-api + ${jakarta.servlet-api.version} + org.assertj assertj-core ${assertj.version} test + + org.mockito + mockito-core + ${mockito-core.version} + diff --git a/pom.xml b/pom.xml index 14f82f05b0b4..893afd0137a2 100644 --- a/pom.xml +++ b/pom.xml @@ -159,6 +159,7 @@ 4.4.13 1.11.0 + 6.1.0 2.9.2 5.3.4 5.5