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

        THudiTable tHudiTable = new THudiTable();
        tHudiTable.setLocation(getTableLocation());

        // 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());
        }
        tHudiTable.setColumns(tColumns);
        if (!tPartitionColumns.isEmpty()) {
            tHudiTable.setPartition_columns(tPartitionColumns);
        }

        // partitions
        List<PartitionKey> partitionKeys = Lists.newArrayList();
        for (DescriptorTable.ReferencedPartitionInfo partition : partitions) {
            partitionKeys.add(partition.getKey());
        }
        List<RemoteFileInfo> hudiPartitions;
        try {
            GetRemoteFilesParams params = GetRemoteFilesParams.newBuilder().setPartitionKeys(partitionKeys).build();
            hudiPartitions = GlobalStateMgr.getCurrentState().getMetadataMgr()
                    .getRemoteFiles(this, params);
        } catch (StarRocksConnectorException e) {
            LOG.warn("Table {} gets partition info failed.", name, e);
            return null;
        }

        HoodieInstant lastInstant = null;
        for (int i = 0; i < hudiPartitions.size(); i++) {
            DescriptorTable.ReferencedPartitionInfo info = partitions.get(i);
            PartitionKey key = info.getKey();
            long partitionId = info.getId();

            THdfsPartition tPartition = new THdfsPartition();
            tPartition.setFile_format(hudiPartitions.get(i).getFormat().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(hudiPartitions.get(i).getFullPath());
            tPartition.setLocation(tPartitionLocation);
            tHudiTable.putToPartitions(partitionId, tPartition);

            // update lastInstant according to remote file info.
            {
                RemoteFileInfo fileInfo = hudiPartitions.get(i);
                for (RemoteFileDesc desc : fileInfo.getFiles()) {
                    HudiRemoteFileDesc hudiDesc = (HudiRemoteFileDesc) desc;
                    HoodieInstant instant = hudiDesc.getHudiInstant();
                    if (instant == null) {
                        continue;
                    }
                    if (lastInstant == null || instant.compareTo(lastInstant) > 0) {
                        lastInstant = instant;
                    }
                }
            }
        }

        Configuration conf = new Configuration();
        if (!Strings.isNullOrEmpty(catalogName)) {
            GlobalStateMgr.getCurrentState()
                    .getConnectorMgr()
                    .getConnector(catalogName)
                    .getMetadata()
                    .getCloudConfiguration()
                    .applyToConfiguration(conf);
        }

        if (tableType == HudiTableType.MOR) {
            tHudiTable.setInstant_time(lastInstant == null ? "" : lastInstant.getTimestamp());
        }

        tHudiTable.setHive_column_names(hudiProperties.get(HUDI_TABLE_COLUMN_NAMES));
        tHudiTable.setHive_column_types(hudiProperties.get(HUDI_TABLE_COLUMN_TYPES));
        tHudiTable.setInput_format(hudiProperties.get(HUDI_TABLE_INPUT_FOAMT));
        tHudiTable.setSerde_lib(hudiProperties.get(HUDI_TABLE_SERDE_LIB));
        tHudiTable.setTime_zone(TimeUtils.getSessionTimeZone());

        TTableDescriptor tTableDescriptor =
                new TTableDescriptor(id, TTableType.HUDI_TABLE, fullSchema.size(), 0, hiveTableName, hiveDbName);
        tTableDescriptor.setHudiTable(tHudiTable);
        return tTableDescriptor;
    }
@Test
    public void testToThrift(
            @Mocked ConnectorMgr connectorMgr,
            @Mocked CatalogConnector catalogConnector,
            @Mocked ConnectorMetadata connectorMetadata,
            @Mocked HoodieTableMetaClient hoodieTableMetaClient) {
        new Expectations() {
            {
                connectorMgr.getConnector(anyString);
                result = catalogConnector;
            }

            {
                catalogConnector.getMetadata();
                result = connectorMetadata;
            }

            {
                connectorMetadata.getCloudConfiguration();
                result = new CloudConfiguration();
                times = 1;
            }
        };

        List<Column> columns = Lists.newArrayList();
        columns.add(new Column("col1", Type.INT, true));
        columns.add(new Column("col2", Type.INT, true));
        long createTime = System.currentTimeMillis();

        Map<String, String> properties = Maps.newHashMap();
        properties.put("hudi.table.base.path", "hdfs://127.0.0.1:10000/hudi");
        HudiTable.Builder tableBuilder = HudiTable.builder()
                .setId(2)
                .setTableName("table0")
                .setCatalogName("catalog")
                .setHiveDbName("db0")
                .setHiveTableName("table0")
                .setResourceName("catalog")
                .setFullSchema(columns)
                .setPartitionColNames(Lists.newArrayList("col1"))
                .setCreateTime(createTime)
                .setHudiProperties(properties);
        HudiTable table = tableBuilder.build();

        TTableDescriptor tTableDescriptor = table.toThrift(ImmutableList.of());
        Assert.assertEquals("db0", tTableDescriptor.getDbName());
        Assert.assertEquals("table0", tTableDescriptor.getTableName());
    }