@Test public void testCommitMultiInsertOverwriteTable() throws MetaException, IOException {
    DerbyConnectorTestUtility connector = derbyConnectorRule.getConnector();
    MetadataStorageTablesConfig metadataStorageTablesConfig = derbyConnectorRule.metadataTablesConfigSupplier().get();
    LocalFileSystem localFileSystem = FileSystem.getLocal(config);
    druidStorageHandler.preCreateTable(tableMock);
    Path taskDirPath = new Path(tableWorkingPath, druidStorageHandler.makeStagingName());
    HdfsDataSegmentPusherConfig pusherConfig = new HdfsDataSegmentPusherConfig();
    pusherConfig.setStorageDirectory(config.get(String.valueOf(HiveConf.ConfVars.DRUID_SEGMENT_DIRECTORY)));
    DataSegmentPusher
        dataSegmentPusher =
        new HdfsDataSegmentPusher(pusherConfig, config, DruidStorageHandlerUtils.JSON_MAPPER);

    // This create and publish the segment to be overwritten
    List<DataSegment>
        existingSegments =
        Collections.singletonList(createSegment(new Path(taskDirPath, DruidStorageHandlerUtils.INDEX_ZIP).toString(),
            new Interval(100, 150, DateTimeZone.UTC),
            "v0",
            new LinearShardSpec(0)));
    DruidStorageHandlerUtils.publishSegmentsAndCommit(connector,
        metadataStorageTablesConfig,
        DATA_SOURCE_NAME,
        existingSegments,
        true,
        config,
        dataSegmentPusher);
    // Check that there is one datasource with the published segment
    Assert.assertArrayEquals(Lists.newArrayList(DATA_SOURCE_NAME).toArray(),
        Lists.newArrayList(DruidStorageHandlerUtils.getAllDataSourceNames(connector, metadataStorageTablesConfig))
            .toArray());

    // Sequence is the following:
    // 1) INSERT with no segments -> Original segment still present in the datasource
    // 2) INSERT OVERWRITE with no segments -> Datasource is empty
    // 3) INSERT OVERWRITE with no segments -> Datasource is empty
    // 4) INSERT with no segments -> Datasource is empty
    // 5) INSERT with one segment -> Datasource has one segment
    // 6) INSERT OVERWRITE with one segment -> Datasource has one segment
    // 7) INSERT with one segment -> Datasource has two segments
    // 8) INSERT OVERWRITE with no segments -> Datasource is empty

    // We start:
    // #1
    druidStorageHandler.commitInsertTable(tableMock, false);
    Assert.assertArrayEquals(Lists.newArrayList(DATA_SOURCE_NAME).toArray(),
        Lists.newArrayList(DruidStorageHandlerUtils.getAllDataSourceNames(connector, metadataStorageTablesConfig))
            .toArray());
    Assert.assertEquals(1, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #2
    druidStorageHandler.commitInsertTable(tableMock, true);
    Assert.assertEquals(0, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #3
    druidStorageHandler.commitInsertTable(tableMock, true);
    Assert.assertEquals(0, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #4
    druidStorageHandler.commitInsertTable(tableMock, true);
    Assert.assertEquals(0, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #5
    DataSegment
        dataSegment1 =
        createSegment(new Path(taskDirPath, DruidStorageHandlerUtils.INDEX_ZIP).toString(),
            new Interval(180, 250, DateTimeZone.UTC),
            "v1",
            new LinearShardSpec(0));
    Path
        descriptorPath1 =
        DruidStorageHandlerUtils.makeSegmentDescriptorOutputPath(dataSegment1,
            new Path(taskDirPath, DruidStorageHandler.SEGMENTS_DESCRIPTOR_DIR_NAME));
    DruidStorageHandlerUtils.writeSegmentDescriptor(localFileSystem, dataSegment1, descriptorPath1);
    druidStorageHandler.commitInsertTable(tableMock, false);
    Assert.assertArrayEquals(Lists.newArrayList(DATA_SOURCE_NAME).toArray(),
        Lists.newArrayList(DruidStorageHandlerUtils.getAllDataSourceNames(connector, metadataStorageTablesConfig))
            .toArray());
    Assert.assertEquals(1, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #6
    DataSegment
        dataSegment2 =
        createSegment(new Path(taskDirPath, DruidStorageHandlerUtils.INDEX_ZIP).toString(),
            new Interval(200, 250, DateTimeZone.UTC),
            "v1",
            new LinearShardSpec(0));
    Path
        descriptorPath2 =
        DruidStorageHandlerUtils.makeSegmentDescriptorOutputPath(dataSegment2,
            new Path(taskDirPath, DruidStorageHandler.SEGMENTS_DESCRIPTOR_DIR_NAME));
    DruidStorageHandlerUtils.writeSegmentDescriptor(localFileSystem, dataSegment2, descriptorPath2);
    druidStorageHandler.commitInsertTable(tableMock, true);
    Assert.assertArrayEquals(Lists.newArrayList(DATA_SOURCE_NAME).toArray(),
        Lists.newArrayList(DruidStorageHandlerUtils.getAllDataSourceNames(connector, metadataStorageTablesConfig))
            .toArray());
    Assert.assertEquals(1, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #7
    DataSegment
        dataSegment3 =
        createSegment(new Path(taskDirPath, DruidStorageHandlerUtils.INDEX_ZIP).toString(),
            new Interval(100, 200, DateTimeZone.UTC),
            "v1",
            new LinearShardSpec(0));
    Path
        descriptorPath3 =
        DruidStorageHandlerUtils.makeSegmentDescriptorOutputPath(dataSegment3,
            new Path(taskDirPath, DruidStorageHandler.SEGMENTS_DESCRIPTOR_DIR_NAME));
    DruidStorageHandlerUtils.writeSegmentDescriptor(localFileSystem, dataSegment3, descriptorPath3);
    druidStorageHandler.commitInsertTable(tableMock, false);
    Assert.assertArrayEquals(Lists.newArrayList(DATA_SOURCE_NAME).toArray(),
        Lists.newArrayList(DruidStorageHandlerUtils.getAllDataSourceNames(connector, metadataStorageTablesConfig))
            .toArray());
    Assert.assertEquals(2, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());

    // #8
    druidStorageHandler.commitInsertTable(tableMock, true);
    Assert.assertEquals(0, getUsedSegmentsList(connector, metadataStorageTablesConfig).size());
  }
Test class for Druid Storage handler.
druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidStorageHandler.java