From ef821c2a22ffdddc56e0280a33356878abbce04c Mon Sep 17 00:00:00 2001 From: Dmitriy Fingerman Date: Thu, 16 Jul 2026 16:43:05 -0400 Subject: [PATCH 1/2] HIVE-29744: Concurrent INSERT INTO on new dynamic partition causes data loss (non-ACID) --- .../apache/hadoop/hive/ql/metadata/Hive.java | 3 +- .../TestHiveLoadPartitionKeepExisting.java | 225 ++++++++++++++++++ 2 files changed, 227 insertions(+), 1 deletion(-) create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveLoadPartitionKeepExisting.java diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java index 3a1b23e9f179..aa82a370b7a7 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java @@ -2899,7 +2899,8 @@ private Partition loadPartitionInternal(Path loadPath, Table tbl, MapMoveTask calls {@code loadPartition} with {@link LoadFileType#KEEP_EXISTING} after + * {@code INSERT INTO} (append) or {@link LoadFileType#REPLACE_ALL} after {@code INSERT OVERWRITE}. + * These tests exercise that commit path directly — not a full SQL {@code INSERT} statement. + */ +public class TestHiveLoadPartitionKeepExisting { + + private static HiveConf hiveConf; + private static Hive hive; + + private String tableName; + + @BeforeClass + public static void setUpClass() throws Exception { + hiveConf = new HiveConfForTest(TestHiveLoadPartitionKeepExisting.class); + // hive-site.xml defaults to SQLStdHiveAuthorizerFactoryForTest (hive-it-util); use the ql + // module factory so this test runs from the ql module alone (same as TestHive). + hiveConf.setVar(HiveConf.ConfVars.HIVE_AUTHORIZATION_MANAGER, + "org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdHiveAuthorizerFactory"); + hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_IN_TEST, true); + MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.HIVE_IN_TEST, true); + SessionState.start(hiveConf); + hive = Hive.get(hiveConf); + } + + @Before + public void setUp() throws Exception { + tableName = "test_load_partition_keep_existing_" + System.nanoTime(); + createExternalPartitionedTable(tableName); + } + + @After + public void tearDown() throws Exception { + hive.dropTable(Warehouse.DEFAULT_DATABASE_NAME, tableName, true, true, true); + } + + /** + * {@link LoadFileType#KEEP_EXISTING} when partition data exists on disk but HMS has no entry yet. + */ + @Test + public void testKeepExistingAppendsWhenPartitionDirExistsButNotInHms() throws Exception { + Table table = hive.getTable(tableName); + Map partSpec = partitionSpec("2026-06-14"); + Path partPath = new Path(table.getDataLocation(), Warehouse.makePartPath(partSpec)); + FileSystem fs = partPath.getFileSystem(hiveConf); + + Path existingFile = new Path(partPath, "000000_0_writer_a"); + fs.mkdirs(partPath); + try (OutputStream out = fs.create(existingFile)) { + out.write("writer-a\n".getBytes()); + } + + Path staging = createStagingDir(fs, table.getDataLocation(), "writer_b"); + Path stagingFile = new Path(staging, "000000_0_writer_b"); + try (OutputStream out = fs.create(stagingFile)) { + out.write("writer-b\n".getBytes()); + } + + hive.loadPartition(staging, table, partSpec, LoadFileType.KEEP_EXISTING, true, false, + false, false, false, false, null, 0, false, false); + + assertTrue("Existing partition file should survive KEEP_EXISTING loadPartition (append)", + fs.exists(new Path(partPath, "000000_0_writer_a"))); + assertTrue("Staged file should be moved in by KEEP_EXISTING loadPartition", + fs.exists(new Path(partPath, "000000_0_writer_b"))); + } + + @Test + public void testReplaceAllReplacesExistingPartitionData() throws Exception { + Table table = hive.getTable(tableName); + Map partSpec = partitionSpec("2026-06-15"); + Path partPath = new Path(table.getDataLocation(), Warehouse.makePartPath(partSpec)); + FileSystem fs = partPath.getFileSystem(hiveConf); + + fs.mkdirs(partPath); + try (OutputStream out = fs.create(new Path(partPath, "000000_0_old"))) { + out.write("old-data\n".getBytes()); + } + + Path staging = createStagingDir(fs, table.getDataLocation(), "overwrite"); + try (OutputStream out = fs.create(new Path(staging, "000000_0_new"))) { + out.write("new-data\n".getBytes()); + } + + hive.loadPartition(staging, table, partSpec, LoadFileType.REPLACE_ALL, true, false, + false, false, false, false, null, 0, true, false); + + assertFalse("REPLACE_ALL loadPartition should remove prior partition data", + fs.exists(new Path(partPath, "000000_0_old"))); + assertTrue("REPLACE_ALL loadPartition should publish staged data", + fs.exists(new Path(partPath, "000000_0_new"))); + } + + @Test + public void testConcurrentKeepExistingAppendsBothStagedLoads() throws Exception { + Table table = hive.getTable(tableName); + Map partSpec = partitionSpec("2026-06-16"); + FileSystem fs = table.getDataLocation().getFileSystem(hiveConf); + + Path stagingA = createStagingDir(fs, table.getDataLocation(), "writer_a"); + try (OutputStream out = fs.create(new Path(stagingA, "000000_0_writer_a"))) { + out.write("writer-a\n".getBytes()); + } + + Path stagingB = createStagingDir(fs, table.getDataLocation(), "writer_b"); + try (OutputStream out = fs.create(new Path(stagingB, "000000_0_writer_b"))) { + out.write("writer-b\n".getBytes()); + } + + CyclicBarrier barrier = new CyclicBarrier(2); + SessionState parentSession = SessionState.get(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future writerA = executor.submit(() -> { + SessionState.setCurrentSessionState(parentSession); + barrierAwait(barrier); + hive.loadPartition(stagingA, table, partSpec, LoadFileType.KEEP_EXISTING, true, false, + false, false, false, false, null, 0, false, false); + return null; + }); + + Future writerB = executor.submit(() -> { + SessionState.setCurrentSessionState(parentSession); + barrierAwait(barrier); + hive.loadPartition(stagingB, table, partSpec, LoadFileType.KEEP_EXISTING, true, false, + false, false, false, false, null, 1, false, false); + return null; + }); + writerA.get(2, TimeUnit.MINUTES); + writerB.get(2, TimeUnit.MINUTES); + } finally { + executor.shutdownNow(); + } + + Path partPath = new Path(table.getDataLocation(), Warehouse.makePartPath(partSpec)); + assertTrue("First KEEP_EXISTING loadPartition should publish its staged file", + fs.exists(new Path(partPath, "000000_0_writer_a"))); + assertTrue("Second KEEP_EXISTING loadPartition should append its staged file", + fs.exists(new Path(partPath, "000000_0_writer_b"))); + } + + private static void barrierAwait(CyclicBarrier barrier) { + try { + barrier.await(1, TimeUnit.MINUTES); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private void createExternalPartitionedTable(String name) throws HiveException { + hive.dropTable(Warehouse.DEFAULT_DATABASE_NAME, name, true, true, true); + hive.createTable(name, + Arrays.asList("value"), + Arrays.asList("load_date"), + TextInputFormat.class, + HiveIgnoreKeyTextOutputFormat.class); + Table table = hive.getTable(name); + table.setTableType(TableType.EXTERNAL_TABLE); + table.getParameters().put("EXTERNAL", "TRUE"); + table.getParameters().put("transactional", "false"); + hive.alterTable(table, false, null, false); + } + + private static Map partitionSpec(String loadDate) { + Map partSpec = new LinkedHashMap<>(); + partSpec.put("load_date", loadDate); + return partSpec; + } + + private static Path createStagingDir(FileSystem fs, Path tablePath, String suffix) throws Exception { + Path staging = new Path(tablePath, "_staging_" + suffix + "_" + System.nanoTime()); + fs.mkdirs(staging); + return staging; + } +} From 0ef8a9eb0f29117171b55bbce6a8a0a73226762d Mon Sep 17 00:00:00 2001 From: Dmitriy Fingerman Date: Fri, 17 Jul 2026 09:47:06 -0400 Subject: [PATCH 2/2] Code review 07/17. --- ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java index aa82a370b7a7..d7eb7281eccc 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java @@ -2898,9 +2898,8 @@ private Partition loadPartitionInternal(Path loadPath, Table tbl, Map