Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
Original file line number Diff line number Diff line change
Expand Up @@ -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" +
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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}).
*
* <p>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.
*
* <p>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);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe this will create a new RestCatalog Object/HttpClient on every call, curious if this can cause any connection leak?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes - Catalogs.loadTable creates a new REST catalog (and HTTP client) on each call, and we don’t close it on this path today. That’s the same pattern as other Catalogs.loadTable usages in the handler. Here it runs once per getSplits during split planning, not per input split. A reload only happens when shouldReloadForServerSideScanPlanning is true.

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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -157,16 +155,18 @@ private static <T extends Scan<T, FileScanTask, CombinedScanTask>> T applyConfig
@Override
public List<InputSplit> 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));
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
12 changes: 0 additions & 12 deletions iceberg/iceberg-rest-catalog-client/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -75,17 +75,5 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-core</artifactId>
<classifier>tests</classifier>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-api</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</project>
Loading
Loading