class DatabaseShardManager implements ShardManager {

    @Override
    public void createTable(long tableId, List<ColumnInfo> columns, boolean bucketed, OptionalLong temporalColumnId)
    {
        StringJoiner tableColumns = new StringJoiner(",\n  ", "  ", ",\n").setEmptyValue("");

        for (ColumnInfo column : columns) {
            String columnType = sqlColumnType(column.getType());
            if (columnType != null) {
                tableColumns.add(minColumn(column.getColumnId()) + " " + columnType);
                tableColumns.add(maxColumn(column.getColumnId()) + " " + columnType);
            }
        }

        StringJoiner coveringIndexColumns = new StringJoiner(", ");

        // Add the max temporal column first to accelerate queries that usually scan recent data
        temporalColumnId.ifPresent(id -> coveringIndexColumns.add(maxColumn(id)));
        temporalColumnId.ifPresent(id -> coveringIndexColumns.add(minColumn(id)));

        String sql;
        if (bucketed) {
            coveringIndexColumns
                    .add("bucket_number")
                    .add("shard_id")
                    .add("shard_uuid");

            sql = "" +
                    "CREATE TABLE " + shardIndexTable(tableId) + " (\n" +
                    "  shard_id BIGINT NOT NULL,\n" +
                    "  shard_uuid BINARY(16) NOT NULL,\n" +
                    "  bucket_number INT NOT NULL\n," +
                    tableColumns +
                    "  PRIMARY KEY (bucket_number, shard_uuid),\n" +
                    "  UNIQUE (shard_id),\n" +
                    "  UNIQUE (shard_uuid),\n" +
                    "  UNIQUE (" + coveringIndexColumns + ")\n" +
                    ")";
        }
        else {
            coveringIndexColumns
                    .add("node_ids")
                    .add("shard_id")
                    .add("shard_uuid");

            sql = "" +
                    "CREATE TABLE " + shardIndexTable(tableId) + " (\n" +
                    "  shard_id BIGINT NOT NULL,\n" +
                    "  shard_uuid BINARY(16) NOT NULL,\n" +
                    "  node_ids VARBINARY(128) NOT NULL,\n" +
                    tableColumns +
                    "  PRIMARY KEY (node_ids, shard_uuid),\n" +
                    "  UNIQUE (shard_id),\n" +
                    "  UNIQUE (shard_uuid),\n" +
                    "  UNIQUE (" + coveringIndexColumns + ")\n" +
                    ")";
        }

        try (Handle handle = dbi.open()) {
            handle.execute(sql);
        }
        catch (DBIException e) {
            throw metadataError(e);
        }
    }

    @Inject public  DatabaseShardManager(
            @ForMetadata IDBI dbi,
            DaoSupplier<ShardDao> shardDaoSupplier,
            NodeSupplier nodeSupplier,
            AssignmentLimiter assignmentLimiter,
            Ticker ticker,
            MetadataConfig config);
    public  DatabaseShardManager(
            IDBI dbi,
            DaoSupplier<ShardDao> shardDaoSupplier,
            NodeSupplier nodeSupplier,
            AssignmentLimiter assignmentLimiter,
            Ticker ticker,
            Duration startupGracePeriod);

}