@Override
  public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
    if (!tableExists(tablePath)) {
      throw new TableNotExistException(getName(), tablePath);
    }

    final String path = inferTablePath(catalogPathStr, tablePath);
    Map<String, String> options = TableOptionProperties.loadFromProperties(path, hadoopConf);
    final Schema latestSchema = getLatestTableSchema(path);
    if (latestSchema != null) {
      List<String> pkColumns = TableOptionProperties.getPkColumns(options);
      // if the table is initialized from spark, the write schema is nullable for pk columns.
      DataType tableDataType = DataTypeUtils.ensureColumnsAsNonNullable(
          AvroSchemaConverter.convertToDataType(latestSchema), pkColumns);
      org.apache.flink.table.api.Schema.Builder builder = org.apache.flink.table.api.Schema.newBuilder()
          .fromRowDataType(tableDataType);
      final String pkConstraintName = TableOptionProperties.getPkConstraintName(options);
      if (!StringUtils.isNullOrEmpty(pkConstraintName)) {
        builder.primaryKeyNamed(pkConstraintName, pkColumns);
      } else if (!CollectionUtils.isNullOrEmpty(pkColumns)) {
        builder.primaryKey(pkColumns);
      }
      final org.apache.flink.table.api.Schema schema = builder.build();
      return CatalogTable.of(
          schema,
          TableOptionProperties.getComment(options),
          TableOptionProperties.getPartitionColumns(options),
          TableOptionProperties.getTableOptions(options));
    } else {
      throw new TableNotExistException(getName(), tablePath);
    }
  }
@Test
  public void testGetTable() throws Exception {
    ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
    // create table
    catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);

    Map<String, String> expectedOptions = new HashMap<>(EXPECTED_OPTIONS);
    expectedOptions.put(FlinkOptions.TABLE_TYPE.key(), FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
    expectedOptions.put(FlinkOptions.INDEX_GLOBAL_ENABLED.key(), "false");
    expectedOptions.put(FlinkOptions.PRE_COMBINE.key(), "true");
    expectedOptions.put("connector", "hudi");
    expectedOptions.put(
        FlinkOptions.PATH.key(),
        String.format("%s/%s/%s", tempFile.getAbsolutePath(), tablePath.getDatabaseName(), tablePath.getObjectName()));

    // test get table
    CatalogBaseTable actualTable = catalog.getTable(tablePath);
    // validate schema
    Schema actualSchema = actualTable.getUnresolvedSchema();
    Schema expectedSchema = Schema.newBuilder().fromResolvedSchema(EXPECTED_TABLE_SCHEMA).build();
    assertEquals(expectedSchema, actualSchema);
    // validate options
    Map<String, String> actualOptions = actualTable.getOptions();
    assertEquals(expectedOptions, actualOptions);
    // validate comment
    assertEquals(EXPECTED_CATALOG_TABLE.getComment(), actualTable.getComment());
    // validate partition key
    assertEquals(EXPECTED_CATALOG_TABLE.getPartitionKeys(), ((CatalogTable) actualTable).getPartitionKeys());
  }