blueflood-core/src/main/java/com/rackspacecloud/blueflood/io/astyanax/AstyanaxReader.java
public <T extends Rollup> Points<T> getDataToRoll(Class<T> type, final Locator locator, Range range, ColumnFamily<Locator, Long> cf) throws IOException {
        AbstractSerializer serializer = Serializers.serializerFor(type);
        // special cases. :( the problem here is that the normal full res serializer returns Number instances instead of
        // SimpleNumber instances.
        // todo: this logic will only become more complicated. It needs to be in its own method and the serializer needs
        // to be known before we ever get to this method (see above comment).
        if (cf == CassandraModel.CF_METRICS_FULL) {
            serializer = Serializers.simpleNumberSerializer;
        } else if ( cf == CassandraModel.CF_METRICS_PREAGGREGATED_FULL) {
            // consider a method for this.  getSerializer(CF, TYPE);
            if (type.equals(BluefloodTimerRollup.class)) {
                serializer = Serializers.timerRollupInstance;
            } else if (type.equals(BluefloodSetRollup.class)) {
                serializer = Serializers.setRollupInstance;
            } else if (type.equals(BluefloodGaugeRollup.class)) {
                serializer = Serializers.gaugeRollupInstance;
            } else if (type.equals(BluefloodCounterRollup.class)) {
                serializer = Serializers.counterRollupInstance;
            }
            else {
                serializer = Serializers.simpleNumberSerializer;
            }
        }

        ColumnList<Long> cols = getColumnsFromDB(locator, cf, range);
        Points<T> points = new Points<T>();
        try {
            for (Column<Long> col : cols) {
                points.add(new Points.Point<T>(col.getName(), (T)col.getValue(serializer)));
            }

            // we only want to count the number of points we
            // get when we're querying the metrics_full
            // we don't do for aggregated or other granularities
            if ( cf == CassandraModel.CF_METRICS_FULL ) {
                Instrumentation.getRawPointsIn5MinHistogram().update(points.getPoints().size());
            }
        } catch (RuntimeException ex) {
            log.error("Problem deserializing data for " + locator + " (" + range + ") from " + cf.getName(), ex);
            throw new IOException(ex);
        }
        return points;
    }
public <T extends Rollup> Points<T> getDataToRoll(Class<T> type, final Locator locator, Range range, ColumnFamily<Locator, Long> cf) throws IOException {
        AbstractSerializer serializer = Serializers.serializerFor(type);
        // special cases. :( the problem here is that the normal full res serializer returns Number instances instead of
        // SimpleNumber instances.
        // todo: this logic will only become more complicated. It needs to be in its own method and the serializer needs
        // to be known before we ever get to this method (see above comment).
        if (cf == CassandraModel.CF_METRICS_FULL) {
            serializer = Serializers.simpleNumberSerializer;
        } else if ( cf == CassandraModel.CF_METRICS_PREAGGREGATED_FULL) {
            // consider a method for this.  getSerializer(CF, TYPE);
            if (type.equals(BluefloodTimerRollup.class)) {
                serializer = Serializers.timerRollupInstance;
            } else if (type.equals(BluefloodSetRollup.class)) {
                serializer = Serializers.setRollupInstance;
            } else if (type.equals(BluefloodGaugeRollup.class)) {
                serializer = Serializers.gaugeRollupInstance;
            } else if (type.equals(BluefloodCounterRollup.class)) {
                serializer = Serializers.counterRollupInstance;
            }
            else {
                serializer = Serializers.simpleNumberSerializer;
            }
        }

        ColumnList<Long> cols = getColumnsFromDB(locator, cf, range);
        Points<T> points = new Points<T>();
        try {
            for (Column<Long> col : cols) {
                points.add(new Points.Point<T>(col.getName(), (T)col.getValue(serializer)));
            }

            // we only want to count the number of points we
            // get when we're querying the metrics_full
            // we don't do for aggregated or other granularities
            if ( cf == CassandraModel.CF_METRICS_FULL ) {
                Instrumentation.getRawPointsIn5MinHistogram().update(points.getPoints().size());
            }
        } catch (RuntimeException ex) {
            log.error("Problem deserializing data for " + locator + " (" + range + ") from " + cf.getName(), ex);
            throw new IOException(ex);
        }
        return points;
    }
todo: A better interface may be to pass the serializer in instead of the class type.