Skip to content

HIVE-29400: Execute Hive Iceberg test-suite agains V3 format #6263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
deniskuzZ merged 4 commits into from
Jan 15, 2026
Merged

HIVE-29400: Execute Hive Iceberg test-suite agains V3 format #6263

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a9e2416
HIVE-29400: Execute Hive Iceberg test-suite agains V3 format
deniskuzZ Jan 10, 2026
56eafae
util classes consolidation
deniskuzZ Jan 12, 2026
89af3e0
review comments
deniskuzZ Jan 13, 2026
47229dc
refactor #2
deniskuzZ Jan 14, 2026
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
4 changes: 1 addition & 3 deletions .../iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,6 @@
import org.apache.iceberg.mr.hive.compaction.IcebergCompactionUtil;
import org.apache.iceberg.mr.hive.writer.HiveIcebergWriter;
import org.apache.iceberg.mr.hive.writer.WriterRegistry;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.base.Splitter;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
Expand Down Expand Up @@ -795,8 +794,7 @@ private static FilesForCommit collectResults(int numTasks, ExecutorService execu
* @param jobId The JobID for the task
* @return The file to store the results
*/
@VisibleForTesting
static String generateJobLocation(String location, Configuration conf, JobID jobId) {
private static String generateJobLocation(String location, Configuration conf, JobID jobId) {
String queryId = conf.get(HiveConf.ConfVars.HIVE_QUERY_ID.varname);
return location + "/temp/" + queryId + "-" + jobId;
}
Expand Down
53 changes: 0 additions & 53 deletions ...eberg-handler/src/test/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandlerStub.java
View file
Open in desktop

This file was deleted.

112 changes: 103 additions & 9 deletions ...ler/src/test/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandlerWithEngineBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import org.apache.hadoop.hive.common.StatsSetupConst;
Expand All @@ -33,14 +34,23 @@
import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.hadoop.ConfigProperties;
import org.apache.iceberg.mr.TestHelper;
import org.apache.iceberg.mr.hive.TestTables.TestTableType;
import org.apache.iceberg.mr.hive.test.TestHiveShell;
import org.apache.iceberg.mr.hive.test.TestTables;
import org.apache.iceberg.mr.hive.test.TestTables.TestTableType;
import org.apache.iceberg.mr.hive.test.concurrent.HiveIcebergStorageHandlerStub;
import org.apache.iceberg.mr.hive.test.concurrent.TestUtilPhaser;
import org.apache.iceberg.mr.hive.test.concurrent.WithMockedStorageHandler;
import org.apache.iceberg.mr.hive.test.utils.HiveIcebergStorageHandlerTestUtils;
import org.apache.iceberg.relocated.com.google.common.base.Throwables;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Tasks;
import org.apache.thrift.TException;
import org.junit.After;
import org.junit.AfterClass;
Expand All @@ -52,9 +62,12 @@
import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.iceberg.mr.hive.TestTables.ALL_TABLE_TYPES;
import static org.apache.iceberg.mr.hive.TestTables.TestTableType.HIVE_CATALOG;
import static org.apache.iceberg.mr.hive.test.TestTables.ALL_TABLE_TYPES;
import static org.apache.iceberg.mr.hive.test.TestTables.TestTableType.HIVE_CATALOG;
import static org.apache.iceberg.mr.hive.test.utils.HiveIcebergStorageHandlerTestUtils.init;
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.junit.Assume.assumeTrue;
Expand All @@ -64,6 +77,8 @@
@RunWith(Parameterized.class)
public abstract class HiveIcebergStorageHandlerWithEngineBase {

private static final Logger LOG = LoggerFactory.getLogger(HiveIcebergStorageHandlerWithEngineBase.class);

public static final String RETRY_STRATEGIES =
"overlay,reoptimize,reexecute_lost_am,dagsubmit,recompile_without_cbo,write_conflict";

Expand Down Expand Up @@ -114,7 +129,7 @@ public static Collection<Object[]> parameters() {

// Run tests with every FileFormat for a single Catalog (HiveCatalog)
for (FileFormat fileFormat : HiveIcebergStorageHandlerTestUtils.FILE_FORMATS) {
IntStream.of(2, 1).forEach(formatVersion -> {
IntStream.rangeClosed(1, 3).forEach(formatVersion -> {
testParams.add(new Object[]{fileFormat, HIVE_CATALOG, false, formatVersion});
// test for vectorization=ON in case of ORC and PARQUET format
if (fileFormat != FileFormat.METADATA) {
Expand Down Expand Up @@ -155,6 +170,9 @@ public static Collection<Object[]> parameters() {
@Rule
public Timeout timeout = new Timeout(500_000, TimeUnit.MILLISECONDS);

@Rule
public WithMockedStorageHandler.Rule mockedStorageHandlerRule = new WithMockedStorageHandler.Rule();

@BeforeClass
public static void beforeClass() {
shell = HiveIcebergStorageHandlerTestUtils.shell();
Expand All @@ -173,11 +191,8 @@ public void before() throws IOException {
HiveConf.setBoolVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, isVectorized);
// Fetch task conversion might kick in for certain queries preventing vectorization code path to be used, so
// we turn it off explicitly to achieve better coverage.
if (isVectorized) {
HiveConf.setVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_FETCH_TASK_CONVERSION, "none");
} else {
HiveConf.setVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_FETCH_TASK_CONVERSION, "more");
}
HiveConf.setVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_FETCH_TASK_CONVERSION,
isVectorized ? "none" : "more");
}

protected void validateTestParams() {
Expand Down Expand Up @@ -210,4 +225,83 @@ protected void validateBasicStats(Table icebergTable, String dbName, String tabl
Assert.assertEquals(summary.get(entry.getValue()), hmsParams.get(entry.getKey()));
}
}

/**
* Executes multiple SQL queries concurrently with controlled synchronization for testing.
*
* <p>This method supports two synchronization modes:
* <ul>
* <li><b>Ext-locking mode</b> ({@code useExtLocking=true}): Queries execute in strict sequential
* order (sql[0], then sql[1], then sql[2], ...) to verify that external table locking prevents
* concurrent execution. Each query waits for its turn before starting.</li>
* <li><b>Barrier synchronization mode</b> ({@code useExtLocking=false}): Queries execute concurrently,
* then commit in order to test optimistic concurrency control and retry on write conflicts.</li>
* </ul>
*
* <p>Uses {@link TestUtilPhaser} and {@link HiveIcebergStorageHandlerStub} to coordinate thread
* execution order deterministically.
*
* @param useExtLocking if true, enables external locking mode with sequential execution;
* if false, enables concurrent execution with ordered commits
* @param retryStrategies comma-separated list of Hive query retry strategies to enable
* @param sql array of SQL queries to execute. If single query provided, it's executed by all threads.
* If multiple queries provided, each thread executes sql[i] where i is the thread index.
* @throws Exception if any query execution fails
*/
protected void executeConcurrently(
boolean useExtLocking, String retryStrategies, String... sql) throws Exception {

int nThreads = sql.length > 1 ? sql.length : 2;
TestUtilPhaser testUtilPhaser = TestUtilPhaser.getInstance();

try {
Tasks.range(nThreads)
.executeWith(Executors.newFixedThreadPool(nThreads))

@Aggarwal-Raghav Aggarwal-Raghav Jan 13, 2026

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.

Executor Service in executeWith(Executors.newFixedThreadPool(nThreads) ) is not closed/shutdown?

@deniskuzZ deniskuzZ Jan 13, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

good point! it was the original code

.run(i -> {
LOG.debug("Thread {} started for query index {}", Thread.currentThread().getName(), i);

TestUtilPhaser.ThreadContext.setQueryIndex(i);

if (useExtLocking) {
TestUtilPhaser.ThreadContext.setUseExtLocking(true);
} else {
testUtilPhaser.register();
}

init(shell, testTables, temp);
HiveConf.setBoolVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, isVectorized);
HiveConf.setVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_FETCH_TASK_CONVERSION, "none");
HiveConf.setBoolVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_OPTIMIZE_METADATA_DELETE, false);
HiveConf.setVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES,
retryStrategies);

if (useExtLocking) {
HiveConf.setBoolVar(shell.getHiveConf(), HiveConf.ConfVars.HIVE_TXN_EXT_LOCKING_ENABLED, true);
shell.getHiveConf().setBoolean(ConfigProperties.LOCK_HIVE_ENABLED, false);

// Wait for turn: sql[0] waits for phase 0 -> 1, sql[1] for 1 -> 2, etc.
// Phase advances when previous query validates snapshot.
LOG.debug("Thread {} (queryIndex={}) waiting for turn",
Thread.currentThread().getName(), i);
testUtilPhaser.awaitTurn();
}

try {
shell.executeStatement(sql.length > 1 ? sql[i] : sql[0]);
LOG.debug("Thread {} (queryIndex={}) completed statement execution",
Thread.currentThread().getName(), i);
} finally {
shell.closeSession();
}
});
} catch (Exception ex) {
Throwable root = Throwables.getRootCause(ex);
if (root instanceof Exception) {
throw (Exception) root;
}
throw new RuntimeException(root);
} finally {
TestUtilPhaser.destroyInstance();
}
}
}
Loading