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..45c9d6c419b6 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_SERVER_SIDE_SCAN_PLANNING_ENABLED( + "hive.iceberg.rest.server.side.scan.planning.enabled", false, + "When true, Hive honors Iceberg REST catalog scan-planning-mode=server for split generation: catalog\n" + + "settings are propagated to Tez/MR jobs and executors reload a REST table so planning can use POST /plan.\n" + + "When false (default), split generation uses the serialized table snapshot even if the catalog requests\n" + + "server-side scan planning."), 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..132b98e7071f 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 @@ -178,6 +178,7 @@ import org.apache.iceberg.hive.HiveTableOperations; import org.apache.iceberg.hive.IcebergCatalogProperties; import org.apache.iceberg.hive.MetastoreUtil; +import org.apache.iceberg.hive.rest.catalog.RestCatalogScanPlanning; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.mr.Catalogs; import org.apache.iceberg.mr.InputFormatConfig; @@ -343,6 +344,10 @@ public void commitJob(JobContext originalContext) { public void configureJobConf(TableDesc tableDesc, JobConf jobConf) { setCommonJobConf(jobConf); configureOutputTableJobConf(tableDesc, jobConf); + if (tableDesc != null && tableDesc.getProperties() != null) { + String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME); + RestCatalogScanPlanning.propagateCatalogPropertiesToJob(conf, catalogName, jobConf); + } if (IcebergVendedCredentialUtil.requestsVendedCredentials(tableDesc.getProperties(), conf)) { IcebergVendedCredentialUtil.refreshVendedCredentialsIfMissing(tableDesc, jobConf, conf); } @@ -1772,6 +1777,9 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD } props.put(InputFormatConfig.PARTITION_SPEC, PartitionSpecParser.toJson(spec)); + String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME); + RestCatalogScanPlanning.propagateCatalogPropertiesToJob(configuration, catalogName, map); + // We need to remove this otherwise the job.xml will be invalid as column comments are separated with '\0' and // the serialization utils fail to serialize this character map.remove("columns.comments"); diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java index 94e7217cefc4..5a60e01910a9 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java @@ -53,6 +53,7 @@ import org.apache.hadoop.mapreduce.JobID; import org.apache.iceberg.AppendFiles; import org.apache.iceberg.BaseTable; +import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFiles; import org.apache.iceberg.MetricsConfig; @@ -69,6 +70,8 @@ import org.apache.iceberg.hadoop.HadoopConfigurable; import org.apache.iceberg.hadoop.HadoopFileIO; import org.apache.iceberg.hadoop.Util; +import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.hive.rest.catalog.RestCatalogScanPlanning; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.mapping.NameMapping; @@ -259,6 +262,57 @@ public static Table deserializeTable(Configuration config, String name) { return table; } + /** + * Resolves the Iceberg {@link Table} for split generation ({@code IcebergInputFormat#getSplits}). + * + *

Serialized 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, catalogNameForScan(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 = catalogNameForScan(conf); + if (StringUtils.isBlank(catalogName)) { + return false; + } + if (!CatalogUtil.ICEBERG_CATALOG_TYPE_REST.equals( + IcebergCatalogProperties.getCatalogType(conf, catalogName))) { + return false; + } + return RestCatalogScanPlanning.requestsServerSidePlanning(catalogName, conf); + } + + private static String catalogNameForScan(Configuration conf) { + String catalogFromTable = conf.get(InputFormatConfig.CATALOG_NAME); + if (StringUtils.isNotBlank(catalogFromTable)) { + return catalogFromTable; + } + 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/mapreduce/IcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java index 89f888255b89..37778e3b3a0f 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); + String executorTableId = tableIdentifier != null ? tableIdentifier : table.name(); + if (tableIdentifier == null) { + conf.set(InputFormatConfig.TABLE_IDENTIFIER, executorTableId); + } + if (conf.get(InputFormatConfig.SERIALIZED_TABLE_PREFIX + executorTableId) == 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(InputFormatConfig.SERIALIZED_TABLE_PREFIX + executorTableId, + 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/rest/TestHiveIcebergServerSideScanPlanning.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanning.java new file mode 100644 index 000000000000..b4b3d344f3fd --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanning.java @@ -0,0 +1,106 @@ +/* + * 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.conf.HiveConf; +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.hive.rest.catalog.RestCatalogScanPlanning; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.hive.HiveTableUtil; +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 Hive server-side REST catalog scan planning via + * {@link HiveTableUtil#resolveTableForScanPlanning}. Embedded REST server coverage is in + * {@code TestHiveIcebergServerSideScanPlanningServerIT} in {@code itests/hive-iceberg-rest-server}. + */ +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) { + RestCatalogScanPlanning.setScanPlanningMode(conf, CATALOG_NAME, "server"); + HiveConf.setBoolVar( + conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED, true); + } + 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/iceberg/iceberg-rest-catalog-client/src/main/java/org/apache/iceberg/hive/rest/catalog/RestCatalogScanPlanning.java b/iceberg/iceberg-rest-catalog-client/src/main/java/org/apache/iceberg/hive/rest/catalog/RestCatalogScanPlanning.java new file mode 100644 index 000000000000..e3d064de853d --- /dev/null +++ b/iceberg/iceberg-rest-catalog-client/src/main/java/org/apache/iceberg/hive/rest/catalog/RestCatalogScanPlanning.java @@ -0,0 +1,201 @@ +/* + * 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 + * limitations under the License. + */ + +package org.apache.iceberg.hive.rest.catalog; + +import java.util.Map; +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 org.apache.iceberg.mr.hive.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_SERVER_SIDE_SCAN_PLANNING_ENABLED} is true. + * Operators can use this helper or set catalog {@code scan-planning-mode} directly in {@code hive-site.xml}. + * + *

Tests: {@code TestRestCatalogScanPlanning} in {@code iceberg-rest-catalog-client}; + * {@code TestHiveIcebergServerSideScanPlanning} in {@code iceberg-handler}; embedded REST server + * tests {@code TestRestCatalogScanPlanningServerIT} and {@code TestHiveIcebergServerSideScanPlanningServerIT} + * in {@code itests/hive-iceberg-rest-server}. + * + * @see REST catalog properties + */ +public final class RestCatalogScanPlanning { + + private RestCatalogScanPlanning() { + } + + public static String catalogPropertyKey(String catalogName) { + return IcebergCatalogProperties.catalogPropertyConfigKey( + catalogName, RESTCatalogProperties.SCAN_PLANNING_MODE); + } + + public static void setScanPlanningMode( + Configuration conf, String catalogName, RESTCatalogProperties.ScanPlanningMode mode) { + conf.set(catalogPropertyKey(catalogName), mode.modeName()); + } + + public static void setScanPlanningMode(Configuration conf, String catalogName, String mode) { + setScanPlanningMode(conf, catalogName, RESTCatalogProperties.ScanPlanningMode.fromString(mode)); + } + + public static RESTCatalogProperties.ScanPlanningMode getScanPlanningMode( + Configuration conf, String catalogName) { + String mode = conf.get( + catalogPropertyKey(catalogName), RESTCatalogProperties.SCAN_PLANNING_MODE_DEFAULT.modeName()); + return RESTCatalogProperties.ScanPlanningMode.fromString(mode); + } + + public static boolean isServerMode(Configuration conf, String catalogName) { + return getScanPlanningMode(conf, catalogName) == RESTCatalogProperties.ScanPlanningMode.SERVER; + } + + /** + * Returns true when Hive server-side REST scan planning is enabled in configuration. + */ + public static boolean isHiveServerSideScanPlanningEnabled(Configuration conf) { + if (conf == null) { + return false; + } + return HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED); + } + + /** + * Returns true when the catalog is configured for server-side scan planning and the Hive feature flag is on. + */ + public static boolean requestsServerSidePlanning(String catalogName, Configuration conf) { + if (conf == null || StringUtils.isEmpty(catalogName)) { + return false; + } + return isHiveServerSideScanPlanningEnabled(conf) && isServerMode(conf, catalogName); + } + + /** + * Returns true when catalog properties should be copied into the Tez/MR job configuration so + * executors can reload a live REST catalog table for server-side scan planning. + */ + public static boolean shouldPropagateCatalogPropertiesToJob(String catalogName, Configuration conf) { + String resolvedCatalogName = 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 requestsServerSidePlanning(resolvedCatalogName, conf); + } + + /** + * Resolves the catalog name from per-table {@code iceberg.catalog} or the session default catalog. + */ + public static String resolveCatalogName(Configuration conf, String catalogNameFromTable) { + if (StringUtils.isNotBlank(catalogNameFromTable)) { + return catalogNameFromTable; + } + if (conf == null) { + return null; + } + return IcebergCatalogProperties.getCatalogName(conf); + } + + /** + * Copies {@code iceberg.catalog..*} entries from the HS2 session configuration into Tez/MR + * 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; + } + + if (!RestCatalogScanPlanning.shouldPropagateCatalogPropertiesToJob(catalogName, sessionConf)) { + return; + } + + propagateCatalogProperties(sessionConf, catalogName, (key, value) -> jobProperties.putIfAbsent(key, value)); + } + + /** + * 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; + } + propagateCatalogProperties(sessionConf, catalogName, (key, value) -> { + if (jobConf.get(key) == null) { + jobConf.set(key, value); + } + }); + } + + private static void propagateCatalogProperties( + Configuration sessionConf, String catalogName, PropertyConsumer consumer) { + String resolvedCatalogName = resolveCatalogName(sessionConf, catalogName); + if (StringUtils.isEmpty(resolvedCatalogName)) { + return; + } + + if (!shouldPropagateCatalogPropertiesToJob(resolvedCatalogName, sessionConf)) { + return; + } + + consumer.accept( + HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED.varname, + String.valueOf(isHiveServerSideScanPlanningEnabled(sessionConf))); + + String sessionDefaultCatalog = + MetastoreConf.getVar(sessionConf, MetastoreConf.ConfVars.CATALOG_DEFAULT); + if (StringUtils.isNotBlank(sessionDefaultCatalog)) { + consumer.accept(MetastoreConf.ConfVars.CATALOG_DEFAULT.getVarname(), sessionDefaultCatalog); + } + + String catalogPrefix = + IcebergCatalogProperties.CATALOG_CONFIG_PREFIX + resolvedCatalogName + "."; + sessionConf.forEach( + entry -> { + if (entry.getKey().startsWith(catalogPrefix)) { + consumer.accept(entry.getKey(), entry.getValue()); + } + }); + } + + private interface PropertyConsumer { + void accept(String key, String value); + } +} diff --git a/iceberg/iceberg-rest-catalog-client/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanning.java b/iceberg/iceberg-rest-catalog-client/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanning.java new file mode 100644 index 000000000000..6d9f3e9af3a9 --- /dev/null +++ b/iceberg/iceberg-rest-catalog-client/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanning.java @@ -0,0 +1,197 @@ +/* + * 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.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.hive.rest.catalog.RestCatalogScanPlanning; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link RestCatalogScanPlanning}: 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 TestRestCatalogScanPlanning { + + /** + * {@link RestCatalogScanPlanning#catalogPropertyKey} 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 catalogPropertyKeyUsesIcebergPropertyName() { + assertThat(RestCatalogScanPlanning.catalogPropertyKey("ice01")) + .isEqualTo( + IcebergCatalogProperties.catalogPropertyConfigKey( + "ice01", RESTCatalogProperties.SCAN_PLANNING_MODE)); + } + + /** + * {@code scan-planning-mode=server} alone is not enough; {@link HiveConf.ConfVars + * #HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED} must also be true. + */ + @Test + void requestsServerSidePlanningFromConfiguration() { + Configuration conf = new Configuration(); + assertThat(RestCatalogScanPlanning.requestsServerSidePlanning("ice01", conf)).isFalse(); + + RestCatalogScanPlanning.setScanPlanningMode(conf, "ice01", "server"); + assertThat(RestCatalogScanPlanning.requestsServerSidePlanning("ice01", conf)).isFalse(); + assertThat(RestCatalogScanPlanning.isServerMode(conf, "ice01")).isTrue(); + + enableHiveServerSideScanPlanning(conf); + assertThat(RestCatalogScanPlanning.requestsServerSidePlanning("ice01", conf)).isTrue(); + assertThat(RestCatalogScanPlanning.getScanPlanningMode(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(RestCatalogScanPlanning.resolveCatalogName(conf, null)).isEqualTo("ice01"); + assertThat(RestCatalogScanPlanning.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(RestCatalogScanPlanning.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isFalse(); + + RestCatalogScanPlanning.setScanPlanningMode(conf, "ice01", "server"); + assertThat(RestCatalogScanPlanning.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isFalse(); + enableHiveServerSideScanPlanning(conf); + assertThat(RestCatalogScanPlanning.shouldPropagateCatalogPropertiesToJob("ice01", conf)).isTrue(); + + Configuration hiveConf = new Configuration(); + hiveConf.set( + IcebergCatalogProperties.catalogPropertyConfigKey("ice01", CatalogUtil.ICEBERG_CATALOG_TYPE), + CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE); + RestCatalogScanPlanning.setScanPlanningMode(hiveConf, "ice01", "server"); + enableHiveServerSideScanPlanning(hiveConf); + assertThat(RestCatalogScanPlanning.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"); + RestCatalogScanPlanning.setScanPlanningMode(sessionConf, "ice01", "server"); + enableHiveServerSideScanPlanning(sessionConf); + sessionConf.set("unrelated.key", "skip"); + + Map jobProperties = Maps.newHashMap(); + RestCatalogScanPlanning.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(RestCatalogScanPlanning.catalogPropertyKey("ice01"), "server") + .containsEntry( + HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED.varname, "true") + .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"); + RestCatalogScanPlanning.setScanPlanningMode(sessionConf, "ice01", "server"); + enableHiveServerSideScanPlanning(sessionConf); + + Configuration jobConf = new Configuration(); + RestCatalogScanPlanning.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(RestCatalogScanPlanning.catalogPropertyKey("ice01"))).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(); + RestCatalogScanPlanning.propagateCatalogPropertiesToJob(sessionConf, "ice01", jobProperties); + + assertThat(jobProperties).isEmpty(); + } + + private static void enableHiveServerSideScanPlanning(Configuration conf) { + HiveConf.setBoolVar( + conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED, true); + } +} diff --git a/itests/hive-iceberg-rest-server/pom.xml b/itests/hive-iceberg-rest-server/pom.xml new file mode 100644 index 000000000000..2c3f1aa7e8a2 --- /dev/null +++ b/itests/hive-iceberg-rest-server/pom.xml @@ -0,0 +1,232 @@ + + + + 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 + + + + + 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..a3b48864450f --- /dev/null +++ b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestHiveIcebergServerSideScanPlanningServerIT.java @@ -0,0 +1,213 @@ +/* + * 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.conf.HiveConf; +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.hive.rest.catalog.RestCatalogScanPlanning; +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; + +/** Embedded REST server tests for the Hive executor reload path for server-side scan planning. */ +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); + RestCatalogScanPlanning.setScanPlanningMode(conf, CATALOG_NAME, "server"); + HiveConf.setBoolVar( + conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED, true); + 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); + RestCatalogScanPlanning.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()); + RestCatalogScanPlanning.setScanPlanningMode(sessionConf, CATALOG_NAME, "server"); + HiveConf.setBoolVar( + sessionConf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED, true); + 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..119a64571e5d --- /dev/null +++ b/itests/hive-iceberg-rest-server/src/test/java/org/apache/iceberg/rest/TestRestCatalogScanPlanningServerIT.java @@ -0,0 +1,171 @@ +/* + * 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.conf.HiveConf; +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.hive.rest.catalog.RestCatalogScanPlanning; +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 RestCatalogScanPlanning}: 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); + RestCatalogScanPlanning.setScanPlanningMode(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() throws IOException { + Configuration conf = new Configuration(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CATALOG_DEFAULT, CATALOG_NAME); + RestCatalogScanPlanning.setScanPlanningMode(conf, CATALOG_NAME, "server"); + HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_ICEBERG_REST_SERVER_SIDE_SCAN_PLANNING_ENABLED, true); + assertThat(IcebergCatalogProperties.getCatalogProperties(conf, CATALOG_NAME)) + .containsEntry(RESTCatalogProperties.SCAN_PLANNING_MODE, "server"); + assertThat(RestCatalogScanPlanning.isServerMode(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..f0a34177b8aa 100644 --- a/pom.xml +++ b/pom.xml @@ -159,6 +159,9 @@ 4.4.13 1.11.0 + + 12.1.8 + 6.1.0 2.9.2 5.3.4 5.5