private void testCompactionWithMerge(CompactionType compactionType, boolean isPartitioned, boolean isBucketed,
                                       String partitionName, List<String> bucketName, List<String> deltaDirNames, List<String> compactDirNames,
                                       boolean createTable, boolean dropTable, boolean useFallback) throws Exception {
    conf.setBoolVar(HiveConf.ConfVars.HIVE_MERGE_COMPACTION_ENABLED, !useFallback);
    conf.setBoolVar(HiveConf.ConfVars.COMPACTOR_CRUD_QUERY_BASED, true);
    String dbName = "default";
    String tableName = "testCompaction";
    // Create test table
    TestDataProvider dataProvider = new TestDataProvider();
    if (createTable) {
      dataProvider.createFullAcidTable(tableName, isPartitioned, isBucketed);
    }
    // Find the location of the table
    IMetaStoreClient metaStoreClient = new HiveMetaStoreClient(conf);
    Table table = metaStoreClient.getTable(dbName, tableName);
    FileSystem fs = FileSystem.get(conf);
    // Insert test data into test table with only insert queries
    dataProvider.insertOnlyTestData(tableName, isPartitioned);
    // Get all data before compaction is run
    List<String> expectedData = dataProvider.getAllData(tableName);
    // Verify deltas
    Assert.assertEquals("Delta directories does not match",
            deltaDirNames, CompactorTestUtil.getBaseOrDeltaNames(fs, AcidUtils.deltaFileFilter, table, partitionName));
    // Verify delete delta
    Assert.assertTrue(CompactorTestUtil.getBaseOrDeltaNames(fs, AcidUtils.deleteEventDeltaDirFilter, table,
            partitionName).isEmpty());
    // Run a compaction which uses only merge compactor / query-based compaction
    CompactorFactory compactorFactory = spy(CompactorFactory.getInstance());
    AtomicReference<Compactor> primary = new AtomicReference<>();
    AtomicReference<Compactor> secondary = new AtomicReference<>();
    doAnswer(invocationOnMock -> {
      Object result = invocationOnMock.callRealMethod();
      Assert.assertTrue(result instanceof CompactorPipeline);
      // Use reflection to fetch inner compactors
      CompactorPipeline compactorPipeline = (CompactorPipeline) result;
      Field field = compactorPipeline.getClass().getDeclaredField("compactor");
      field.setAccessible(true);
      Compactor compactor = (Compactor) field.get(compactorPipeline);
      Assert.assertTrue(compactor instanceof CompactorPipeline.FallbackCompactor);
      CompactorPipeline.FallbackCompactor fallbackCompactor = spy((CompactorPipeline.FallbackCompactor) compactor);
      field.set(compactorPipeline, fallbackCompactor);
      field = fallbackCompactor.getClass().getDeclaredField("primaryCompactor");
      field.setAccessible(true);
      Compactor compactor1 = (Compactor) field.get(fallbackCompactor);
      Assert.assertTrue(compactor1 instanceof MergeCompactor);
      compactor1 = spy(compactor1);
      field.set(fallbackCompactor, compactor1);
      field = fallbackCompactor.getClass().getDeclaredField("secondaryCompactor");
      field.setAccessible(true);
      Compactor compactor2 = (Compactor) field.get(fallbackCompactor);
      if (compactionType == CompactionType.MAJOR) {
        Assert.assertTrue(compactor2 instanceof MajorQueryCompactor);
      } else {
        Assert.assertTrue(compactor2 instanceof MinorQueryCompactor);
      }
      compactor2 = spy(compactor2);
      field.set(fallbackCompactor, compactor2);
      primary.set(compactor1);
      secondary.set(compactor2);
      return result;
    }).when(compactorFactory).getCompactorPipeline(any(), any(), any(), any());

    Worker worker = new Worker(compactorFactory);
    worker.setConf(conf);
    worker.init(new AtomicBoolean(true));
    TxnStore txnHandler = TxnUtils.getTxnStore(conf);
    CompactionRequest cr = new CompactionRequest(dbName, tableName, compactionType);
    if (partitionName == null) {
      txnHandler.compact(cr);
      worker.run();
    } else {
      cr.setPartitionname(partitionName);
      txnHandler.compact(cr);
      worker.run();
    }

    if (useFallback) {
      verify(primary.get(), times(1)).run(any());
      verify(secondary.get(), times(1)).run(any());
    } else {
      verify(primary.get(), times(1)).run(any());
      verify(secondary.get(), times(0)).run(any());
    }

    // Clean up resources
    CompactorTestUtil.runCleaner(conf);
    // Only 1 compaction should be in the response queue with succeeded state
    List<ShowCompactResponseElement> compacts =
            TxnUtils.getTxnStore(conf).showCompact(new ShowCompactRequest()).getCompacts();
    compacts.forEach(c -> Assert.assertEquals("succeeded", c.getState()));
    // Verify directories after compaction
    List<String> actualDirAfterComp =
            CompactorTestUtil.getBaseOrDeltaNames(fs, compactionType == CompactionType.MAJOR ? AcidUtils.baseFileFilter :
                    AcidUtils.deltaFileFilter, table, partitionName);
    Assert.assertEquals("Base directory does not match after compaction", compactDirNames, actualDirAfterComp);
    Assert.assertTrue(CompactorTestUtil.getBaseOrDeltaNames(fs, AcidUtils.deleteEventDeltaDirFilter, table,
            partitionName).isEmpty());
    // Verify bucket files in delta dirs
    Assert.assertEquals("Bucket names are not matching after compaction", bucketName,
            CompactorTestUtil.getBucketFileNames(fs, table, partitionName, actualDirAfterComp.get(0)));
    // Verify all contents
    List<String> actualData = dataProvider.getAllData(tableName);
    Assert.assertEquals(expectedData, actualData);
    // Clean up
    if (dropTable) {
      dataProvider.dropTable(tableName);
    }
    conf.setBoolVar(HiveConf.ConfVars.HIVE_MERGE_COMPACTION_ENABLED, false);
  }
Tests whether hive.llap.io.etl.skip.format config is handled properly whenever QueryCompactor#runCompactionQueries
is invoked.
@throws Exception
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCrudCompactorOnTez.java