@Override
    public TTableDescriptor toThrift(List<ReferencedPartitionInfo> partitions) {
        Preconditions.checkNotNull(partitions);

        THdfsTable tHdfsTable = new THdfsTable();
        tHdfsTable.setHdfs_base_dir(tableLocation);

        // columns and partition columns
        Set<String> partitionColumnNames = Sets.newHashSet();
        List<TColumn> tPartitionColumns = Lists.newArrayList();
        List<TColumn> tColumns = Lists.newArrayList();

        for (Column column : getPartitionColumns()) {
            tPartitionColumns.add(column.toThrift());
            partitionColumnNames.add(column.getName());
        }
        for (Column column : getBaseSchema()) {
            if (partitionColumnNames.contains(column.getName())) {
                continue;
            }
            tColumns.add(column.toThrift());
        }
        tHdfsTable.setColumns(tColumns);
        if (!tPartitionColumns.isEmpty()) {
            tHdfsTable.setPartition_columns(tPartitionColumns);
        }

        // partitions
        List<String> partitionNames = Lists.newArrayList();
        for (ReferencedPartitionInfo partition : partitions) {
            partitionNames.add(PartitionUtil.toHivePartitionName(getPartitionColumnNames(), partition.getKey()));
        }
        List<PartitionInfo> hivePartitions;
        try {
            useMetadataCache = true;
            hivePartitions = GlobalStateMgr.getCurrentState().getMetadataMgr()
                    .getPartitions(this.getCatalogName(), this, partitionNames);
        } catch (StarRocksConnectorException e) {
            LOG.warn("table {} gets partition info failed.", name, e);
            return null;
        }

        for (int i = 0; i < hivePartitions.size(); i++) {
            ReferencedPartitionInfo info = partitions.get(i);
            PartitionKey key = info.getKey();
            long partitionId = info.getId();

            THdfsPartition tPartition = new THdfsPartition();
            tPartition.setFile_format(hivePartitions.get(i).getFileFormat().toThrift());

            List<LiteralExpr> keys = key.getKeys();
            tPartition.setPartition_key_exprs(keys.stream().map(Expr::treeToThrift).collect(Collectors.toList()));

            THdfsPartitionLocation tPartitionLocation = new THdfsPartitionLocation();
            tPartitionLocation.setPrefix_index(-1);
            tPartitionLocation.setSuffix(hivePartitions.get(i).getFullPath());
            tPartition.setLocation(tPartitionLocation);
            tHdfsTable.putToPartitions(partitionId, tPartition);
        }

        tHdfsTable.setSerde_lib(hiveProperties.get(HIVE_TABLE_SERDE_LIB));
        tHdfsTable.setInput_format(hiveProperties.get(HIVE_TABLE_INPUT_FORMAT));
        tHdfsTable.setHive_column_names(hiveProperties.get(HIVE_TABLE_COLUMN_NAMES));
        tHdfsTable.setHive_column_types(hiveProperties.get(HIVE_TABLE_COLUMN_TYPES));
        tHdfsTable.setSerde_properties(serdeProperties);
        tHdfsTable.setTime_zone(TimeUtils.getSessionTimeZone());

        TTableDescriptor tTableDescriptor = new TTableDescriptor(id, TTableType.HDFS_TABLE, fullSchema.size(),
                0, hiveTableName, hiveDbName);
        tTableDescriptor.setHdfsTable(tHdfsTable);
        return tTableDescriptor;
    }
@Test
    public void testCreateExternalTableWithStorageFormat(@Mocked MetadataMgr metadataMgr) throws Exception {

        List<String> targetFormats = new ArrayList<>();
        targetFormats.add("AVRO");
        targetFormats.add("RCBINARY");
        targetFormats.add("RCTEXT");
        targetFormats.add("SEQUENCE");

        for (String targetFormat : targetFormats) {
            HiveTable oTable = createExternalTableByFormat(targetFormat);
            String inputFormatClass = HiveStorageFormat.get(targetFormat).getInputFormat();
            String serde = HiveStorageFormat.get(targetFormat).getSerde();

            new Expectations() {
                {
                    GlobalStateMgr.getCurrentState().getMetadataMgr();
                    result = metadataMgr;
                    minTimes = 0;

                    metadataMgr.getTable(anyString, anyString, anyString);
                    result = oTable;
                }
            };

            String createTableSql =
                    "create external table if not exists  db.hive_tbl (col1 int, col2 int) engine=hive properties " +
                            "(\"resource\"=\"hive0\", \"database\"=\"db0\", \"table\"=\"table0\")";
            CreateTableStmt createTableStmt =
                    (CreateTableStmt) UtFrameUtils.parseStmtWithNewParser(createTableSql, connectContext);
            com.starrocks.catalog.Table table = createTable(createTableStmt);

            Assert.assertTrue(table instanceof HiveTable);
            HiveTable hiveTable = (HiveTable) table;
            List<DescriptorTable.ReferencedPartitionInfo> partitions = new ArrayList<>();
            TTableDescriptor tTableDescriptor = hiveTable.toThrift(partitions);

            Assert.assertEquals(tTableDescriptor.getHdfsTable().getInput_format(), inputFormatClass);
            Assert.assertEquals(tTableDescriptor.getHdfsTable().getSerde_lib(), serde);
            Assert.assertEquals(tTableDescriptor.getHdfsTable().getHive_column_names(), "col2");
            Assert.assertEquals(tTableDescriptor.getHdfsTable().getHive_column_types(), "INT");
        }
    }