@Override
    public Table getTable(String dbName, String tblName) {
        if (metastore.isPresent()) {
            return metastore.get().getTable(dbName, tblName);
        }
        String fullTableName = getKuduFullTableName(dbName, tblName);
        try {
            if (!kuduClient.tableExists(fullTableName)) {
                LOG.error("Kudu table {}.{} does not exist.", dbName, tblName);
                return null;
            }
            return tables.computeIfAbsent(fullTableName,
                    f -> {
                        org.apache.kudu.client.KuduTable kuduNativeTable;
                        try {
                            kuduNativeTable = kuduClient.openTable(fullTableName);
                        } catch (KuduException e) {
                            throw new StarRocksConnectorException("Failed to open table %s.", fullTableName, e);
                        }
                        List<ColumnSchema> columnSchemas = kuduNativeTable.getSchema().getColumns();
                        ArrayList<Column> columns = new ArrayList<>(columnSchemas.size());
                        Set<Integer> parColumnIds =
                                Sets.newHashSet(kuduNativeTable.getPartitionSchema().getRangeSchema().getColumnIds());
                        List<String> partColNames = Lists.newArrayList();
                        for (int i = 0; i < columnSchemas.size(); i++) {
                            ColumnSchema columnSchema = columnSchemas.get(i);
                            String fieldName = columnSchema.getName();
                            boolean isKey = columnSchema.isKey();
                            String comment = columnSchema.getComment();
                            Type fieldType = ColumnTypeConverter.fromKuduType(columnSchema);
                            Column column = new Column(fieldName, fieldType, isKey,
                                    null, true, null, comment);
                            columns.add(column);
                            if (parColumnIds.contains(i)) {
                                partColNames.add(fieldName);
                            }
                        }
                        return new KuduTable(masterAddresses, this.catalogName, dbName, tblName, fullTableName,
                                columns, partColNames);
                    });
        } catch (KuduException e) {
            throw new StarRocksConnectorException("Failed to get table %s.", fullTableName, e);
        }
    }
@Test
    public void testGetTable(@Mocked org.apache.kudu.client.KuduTable mockedTable) throws KuduException {
        KuduMetadata metadata = new KuduMetadata(KUDU_CATALOG, new HdfsEnvironment(), KUDU_MASTER, true,
                SCHEMA_EMULATION_PREFIX, Optional.empty());
        new Expectations() {
            {
                client.tableExists(anyString);
                result = true;
                client.openTable(anyString);
                result = mockedTable;
                mockedTable.getSchema();
                result = SCHEMA;
                mockedTable.getPartitionSchema();
                result = EMPTY_PARTITION_SCHEMA;
            }
        };
        Table table = metadata.getTable("db1", "tbl1");
        KuduTable kuduTable = (KuduTable) table;
        Assert.assertEquals("test_kudu_catalog", kuduTable.getCatalogName());
        Assert.assertEquals("db1", kuduTable.getDbName());
        Assert.assertEquals("tbl1", kuduTable.getTableName());
        Assert.assertEquals(2, kuduTable.getColumns().size());
        Assert.assertEquals(0, kuduTable.getPartitionColumnNames().size());
        Assert.assertEquals(ScalarType.INT, kuduTable.getColumns().get(0).getType());
        Assert.assertTrue(kuduTable.getBaseSchema().get(0).isAllowNull());
        Assert.assertEquals(ScalarType.createVarcharType(CATALOG_MAX_VARCHAR_LENGTH),
                kuduTable.getBaseSchema().get(1).getType());
        Assert.assertTrue(kuduTable.getBaseSchema().get(1).isAllowNull());
    }