### Question:
HiveTableSink extends OutputFormatTableSink<Row> implements PartitionableTableSink, OverwritableTableSink { @Override public void setOverwrite(boolean overwrite) { this.overwrite = overwrite; } HiveTableSink(JobConf jobConf, ObjectPath tablePath, CatalogTable table); @Override OutputFormat<Row> getOutputFormat(); @Override TableSink<Row> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes); @Override DataType getConsumedDataType(); @Override TableSchema getTableSchema(); @Override List<String> getPartitionFieldNames(); @Override void setStaticPartition(Map<String, String> partitionSpec); @Override void setOverwrite(boolean overwrite);  }### Answer:
@Test public void testInsertOverwrite() throws Exception { String dbName = "default"; String tblName = "dest"; RowTypeInfo rowTypeInfo = createDestTable(dbName, tblName, 0); ObjectPath tablePath = new ObjectPath(dbName, tblName); TableEnvironment tableEnv = HiveTestUtils.createTableEnv(); List<Row> toWrite = generateRecords(5); Table src = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo)); tableEnv.registerTable("src", src); CatalogTable table = (CatalogTable) hiveCatalog.getTable(tablePath); tableEnv.registerTableSink("destSink", new HiveTableSink(new JobConf(hiveConf), tablePath, table)); tableEnv.sqlQuery("select * from src").insertInto("destSink"); tableEnv.execute("mytest"); verifyWrittenData(toWrite, hiveShell.executeQuery("select * from " + tblName)); toWrite = generateRecords(3); Table src1 = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo)); tableEnv.registerTable("src1", src1); HiveTableSink sink = new HiveTableSink(new JobConf(hiveConf), tablePath, table); sink.setOverwrite(true); tableEnv.registerTableSink("destSink1", sink); tableEnv.sqlQuery("select * from src1").insertInto("destSink1"); tableEnv.execute("mytest"); verifyWrittenData(toWrite, hiveShell.executeQuery("select * from " + tblName)); hiveCatalog.dropTable(tablePath, false); }