### Question:
HiveTableSink extends OutputFormatTableSink<Row> implements PartitionableTableSink, OverwritableTableSink { @Override public void setStaticPartition(Map<String, String> partitionSpec) { staticPartitionSpec = new LinkedHashMap<>(); for (String partitionCol : getPartitionFieldNames()) { if (partitionSpec.containsKey(partitionCol)) { staticPartitionSpec.put(partitionCol, partitionSpec.get(partitionCol)); } } } 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 testInsertIntoStaticPartition() throws Exception { String dbName = "default"; String tblName = "dest"; RowTypeInfo rowTypeInfo = createDestTable(dbName, tblName, 1); ObjectPath tablePath = new ObjectPath(dbName, tblName); TableEnvironment tableEnv = HiveTestUtils.createTableEnv(); List<Row> toWrite = generateRecords(1); Table src = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo)); tableEnv.registerTable("src", src); Map<String, String> partSpec = new HashMap<>(); partSpec.put("s", "a"); CatalogTable table = (CatalogTable) hiveCatalog.getTable(tablePath); HiveTableSink hiveTableSink = new HiveTableSink(new JobConf(hiveConf), tablePath, table); hiveTableSink.setStaticPartition(partSpec); tableEnv.registerTableSink("destSink", hiveTableSink); tableEnv.sqlQuery("select * from src").insertInto("destSink"); tableEnv.execute("mytest"); assertEquals(toWrite.size(), hiveCatalog.listPartitions(tablePath).size()); verifyWrittenData(toWrite, hiveShell.executeQuery("select * from " + tblName)); hiveCatalog.dropTable(tablePath, false); }