public HiveTableSink(HiveTable hiveTable, TupleDescriptor desc, boolean isStaticPartitionSink, SessionVariable sessionVariable) {
        this.desc = desc;
        this.stagingDir = HiveWriteUtils.getStagingDir(hiveTable, sessionVariable.getHiveTempStagingDir());
        this.partitionColNames = hiveTable.getPartitionColumnNames();
        this.dataColNames = hiveTable.getDataColumnNames();
        this.tableIdentifier = hiveTable.getUUID();
        this.isStaticPartitionSink = isStaticPartitionSink;
        HiveStorageFormat format = hiveTable.getStorageFormat();
        if (format != HiveStorageFormat.PARQUET && format != HiveStorageFormat.ORC
                && format != HiveStorageFormat.TEXTFILE) {
            throw new StarRocksConnectorException("Writing to hive table in [%s] format is not supported.", format.name());
        }
        this.fileFormat = hiveTable.getStorageFormat().name().toLowerCase();
        if (format == HiveStorageFormat.TEXTFILE) {
            this.textFileFormatDesc = Optional.of(toTextFileFormatDesc(hiveTable.getSerdeProperties()));
            this.compressionType = String.valueOf(TCompressionType.NO_COMPRESSION);
        } else {
            this.compressionType = sessionVariable.getConnectorSinkCompressionCodec();
        }
        this.targetMaxFileSize = sessionVariable.getConnectorSinkTargetMaxFileSize();
        String catalogName = hiveTable.getCatalogName();
        Connector connector = GlobalStateMgr.getCurrentState().getConnectorMgr().getConnector(catalogName);
        Preconditions.checkState(connector != null,
                String.format("connector of catalog %s should not be null", catalogName));

        this.cloudConfiguration = connector.getMetadata().getCloudConfiguration();

        Preconditions.checkState(cloudConfiguration != null,
                String.format("cloudConfiguration of catalog %s should not be null", catalogName));
    }
@Test
    public void testHiveTableSink(@Mocked CatalogConnector hiveConnector) {
        HiveTable.Builder builder = HiveTable.builder()
                .setId(ConnectorTableId.CONNECTOR_ID_GENERATOR.getNextId().asInt())
                .setTableName("hive_table")
                .setCatalogName("hive_catalog")
                .setResourceName(toResourceName("hive_catalog", "hive"))
                .setHiveDbName("hive_db")
                .setHiveTableName("hive_table")
                .setPartitionColumnNames(Lists.newArrayList("p1"))
                .setDataColumnNames(Lists.newArrayList("c1"))
                .setFullSchema(Lists.newArrayList(new Column("c1", Type.INT), new Column("p1", Type.INT)))
                .setTableLocation("hdfs://hadoop01:9000/tableLocation")
                .setProperties(new HashMap<>())
                .setStorageFormat(HiveStorageFormat.PARQUET)
                .setCreateTime(System.currentTimeMillis());

        new Expectations() {
            {
                hiveConnector.getMetadata().getCloudConfiguration();
                result = CloudConfigurationFactory.buildCloudConfigurationForStorage(new HashMap<>());
                minTimes = 1;
            }
        };

        ConnectorMgr connectorMgr = AnalyzeTestUtil.getConnectContext().getGlobalStateMgr().getConnectorMgr();
        new Expectations(connectorMgr) {
            {
                connectorMgr.getConnector("hive_catalog");
                result = hiveConnector;
                minTimes = 1;
            }
        };

        TupleDescriptor desc = new TupleDescriptor(new TupleId(0));
        HiveTableSink sink = new HiveTableSink(builder.build(), desc, true, new SessionVariable());
        Assert.assertNull(sink.getExchNodeId());
        Assert.assertNull(sink.getOutputPartition());
        Assert.assertNull(sink.getOutputPartition());
        Assert.assertTrue(sink.canUsePipeLine());
        Assert.assertTrue(sink.canUseRuntimeAdaptiveDop());
        Assert.assertTrue(sink.getStagingDir().contains("/tmp/starrocks"));
        Assert.assertTrue(sink.getExplainString("SINK", TExplainLevel.NORMAL).contains(
                "SINKHive TABLE SINK\n" +
                        "SINK  TABLE: hive_catalog.hive_db.hive_table"));
        TDataSink tDataSink = sink.toThrift();
        Assert.assertEquals(TDataSinkType.HIVE_TABLE_SINK, tDataSink.getType());
        THiveTableSink tHiveTableSink = tDataSink.getHive_table_sink();
        Assert.assertTrue(tHiveTableSink.getStaging_dir().startsWith("hdfs://hadoop01:9000/tmp/starrocks"));
        Assert.assertEquals("parquet", tHiveTableSink.getFile_format());
        Assert.assertEquals("c1", tHiveTableSink.getData_column_names().get(0));
        Assert.assertEquals("p1", tHiveTableSink.getPartition_column_names().get(0));
        Assert.assertEquals(TCompressionType.NO_COMPRESSION, tHiveTableSink.getCompression_type());
        Assert.assertTrue(tHiveTableSink.is_static_partition_sink);
        Assert.assertEquals(TCloudType.DEFAULT, tHiveTableSink.getCloud_configuration().cloud_type);

        builder.setStorageFormat(HiveStorageFormat.AVRO);
        ExceptionChecker.expectThrowsWithMsg(StarRocksConnectorException.class,
                "Writing to hive table in [AVRO] format is not supported",
                () ->new HiveTableSink(builder.build(), desc, true, new SessionVariable()));
    }