class ShardedDataTypeHandler extends StatsDEnabledDataTypeHandler<KEYIN> implements DataTypeHandler<KEYIN> {

    @Override
    public void setup(TaskAttemptContext context) {
        markingFunctions = MarkingFunctions.Factory.createMarkingFunctions();
        
        Configuration conf = context.getConfiguration();
        shardIdFactory = new ShardIdFactory(conf);
        
        String tableName = conf.get(SHARD_TNAME, null);
        if (null == tableName)
            log.error(SHARD_TNAME + " not specified, no events will be created, and the global index will be useless");
        else
            setShardTableName(new Text(tableName));
        
        tableName = conf.get(SHARD_STATS_TNAME, null);
        if (null == tableName)
            log.warn(SHARD_STATS_TNAME + " not specified, no global index mutations will be created.");
        else {
            setIndexStatsTableName(new Text(tableName));
            setProduceStats(true);
        }
        
        tableName = conf.get(SHARD_GIDX_TNAME, null);
        if (null == tableName)
            log.warn(SHARD_GIDX_TNAME + " not specified, no global index mutations will be created.");
        else
            setShardIndexTableName(new Text(tableName));
        
        tableName = conf.get(SHARD_GRIDX_TNAME, null);
        if (null == tableName)
            log.warn(SHARD_GRIDX_TNAME + " not specified, no global reverse index mutations will be created.");
        else
            setShardReverseIndexTableName(new Text(tableName));
        
        tableName = conf.get(METADATA_TABLE_NAME, null);
        if (null == tableName)
            log.warn(METADATA_TABLE_NAME + " not specified, no metadata will be created, I hope nothing requires normalizers.");
        else
            setMetadataTableName(new Text(tableName));
        
        tableName = (LoadDateTableConfigHelper.isLoadDatesEnabled(conf) ? LoadDateTableConfigHelper.getLoadDatesTableName(conf) : null);
        if (null == tableName)
            log.warn(LoadDateTableConfigHelper.LOAD_DATES_TABLE_NAME_PROP + " not specified, no load dates will be created");
        else
            setLoadDatesTableName(new Text(tableName));
        
        if (getMetadataTableName() != null) {
            setMetadata(ingestConfig.createMetadata(getShardTableName(), getMetadataTableName(), getLoadDatesTableName(), getShardIndexTableName(),
                            getShardReverseIndexTableName(), conf.getBoolean(METADATA_TERM_FREQUENCY, true)));
        }
        
        tableName = conf.get(SHARD_DINDX_NAME, null);
        if (null == tableName) {
            log.warn(SHARD_DINDX_NAME + " not specified, no term dictionary will be created.");
        } else {
            setShardDictionaryIndexTableName(new Text(tableName));
            this.setupDictionaryCache(conf.getInt(SHARD_DICTIONARY_CACHE_ENTRIES, SHARD_DINDEX_CACHE_DEFAULT_SIZE));
            
        }
        
        setupToReindexIfEnabled(conf);
        
        // enabled by default
        this.bloomFiltersEnabled = conf.getBoolean(SHARD_ININDEX_BLOOM, false);
        if (this.bloomFiltersEnabled) {
            this.bloomFilteringDiskThreshold = conf.getFloat(SHARD_ININDEX_BLOOM_DISK_THRESHOLD, 0.0f);
            this.bloomFilteringDiskThresholdPath = conf.get(SHARD_ININDEX_BLOOM_DISK_THRESHOLD_PATH,
                            DiskSpaceStarvationStrategy.DEFAULT_PATH_FOR_DISK_SPACE_VALIDATION);
            this.bloomFilteringMemoryThreshold = conf.getFloat(SHARD_ININDEX_BLOOM_MEMORY_THRESHOLD, 0.0f);
            this.bloomFilteringTimeoutThreshold = conf.getFloat(SHARD_ININDEX_BLOOM_TIMEOUT_THRESHOLD, 0.0f);
            this.bloomFilteringOptimumMaxFilterSize = conf.getInt(SHARD_ININDEX_BLOOM_OPTIMUM_MAX_FILTER_SIZE, -1);
        }
        
        // Event key suppression
        this.suppressEventKeys = conf.getBoolean(SUPPRESS_EVENT_KEYS, false);
    }

}