### Question:
HBaseMetadataProvider extends AbstractMetadataProvider { @Override public DatasetDescriptor create(String namespace, String name, DatasetDescriptor descriptor) { Preconditions.checkArgument(DEFAULT_NAMESPACE.equals(namespace), "Non-default namespaces are not supported"); Preconditions.checkNotNull(name, "Dataset name cannot be null"); Preconditions.checkNotNull(descriptor, "Descriptor cannot be null"); Compatibility.checkAndWarn( namespace, HBaseMetadataProvider.getTableName(name), descriptor.getSchema()); Preconditions.checkArgument(descriptor.isColumnMapped(), "Cannot create dataset %s: missing column mapping", name); try { String managedSchemaName = "managed_schemas"; if (!hbaseAdmin.tableExists(managedSchemaName)) { HTableDescriptor table = new HTableDescriptor(managedSchemaName); table.addFamily(new HColumnDescriptor("meta")); table.addFamily(new HColumnDescriptor("schema")); table.addFamily(new HColumnDescriptor(Constants.SYS_COL_FAMILY)); hbaseAdmin.createTable(table); } } catch (IOException e) { throw new DatasetIOException("Cannot open schema table", e); } Schema schema = getEmbeddedSchema(descriptor); String entitySchemaString = schema.toString(true); AvroEntitySchema entitySchema = new AvroEntitySchema( schema, entitySchemaString, descriptor.getColumnMapping()); String tableName = getTableName(name); String entityName = getEntityName(name); schemaManager.refreshManagedSchemaCache(tableName, entityName); schemaManager.createSchema(tableName, entityName, entitySchemaString, "org.kitesdk.data.hbase.avro.AvroKeyEntitySchemaParser", "org.kitesdk.data.hbase.avro.AvroKeySerDe", "org.kitesdk.data.hbase.avro.AvroEntitySerDe"); try { if (!hbaseAdmin.tableExists(tableName)) { HTableDescriptor desc = new HTableDescriptor(tableName); Set<String> familiesToAdd = entitySchema.getColumnMappingDescriptor() .getRequiredColumnFamilies(); familiesToAdd.add(new String(Constants.SYS_COL_FAMILY)); familiesToAdd.add(new String(Constants.OBSERVABLE_COL_FAMILY)); for (String columnFamily : familiesToAdd) { desc.addFamily(columnFamily(columnFamily, descriptor)); } hbaseAdmin.createTable(desc); } else { Set<String> familiesToAdd = entitySchema.getColumnMappingDescriptor() .getRequiredColumnFamilies(); familiesToAdd.add(new String(Constants.SYS_COL_FAMILY)); familiesToAdd.add(new String(Constants.OBSERVABLE_COL_FAMILY)); HTableDescriptor desc = hbaseAdmin.getTableDescriptor(tableName .getBytes()); for (HColumnDescriptor columnDesc : desc.getColumnFamilies()) { String familyName = columnDesc.getNameAsString(); if (familiesToAdd.contains(familyName)) { familiesToAdd.remove(familyName); } } if (familiesToAdd.size() > 0) { hbaseAdmin.disableTable(tableName); try { for (String family : familiesToAdd) { hbaseAdmin.addColumn(tableName, columnFamily(family, descriptor)); } } finally { hbaseAdmin.enableTable(tableName); } } } } catch (IOException e) { throw new DatasetIOException("Cannot prepare table: " + name, e); } return getDatasetDescriptor(schema, descriptor.getLocation()); } HBaseMetadataProvider(HBaseAdmin hbaseAdmin, SchemaManager schemaManager); @Override DatasetDescriptor create(String namespace, String name, DatasetDescriptor descriptor); @Override DatasetDescriptor update(String namespace, String name, DatasetDescriptor descriptor); @Override DatasetDescriptor load(String namespace, String name); @Override boolean delete(String namespace, String name); @Override boolean moveToTrash(String namespace, String name); @Override boolean exists(String namespace, String name); @Override Collection<String> namespaces(); @Override Collection<String> datasets(String namespace);  }### Answer:
@Test public void testBasic() { DatasetDescriptor desc = provider.create("default", tableName + ".TestEntity", new DatasetDescriptor.Builder().schemaLiteral(testEntity).build()); ColumnMapping columnMapping = desc.getColumnMapping(); PartitionStrategy partStrat = desc.getPartitionStrategy(); assertEquals(9, columnMapping.getFieldMappings().size()); assertEquals(2, Accessor.getDefault().getFieldPartitioners(partStrat).size()); }