class DataSetService extends BaseDataSetService {

    @RequestMapping(value = "/datasets/{id}/content", method = RequestMethod.GET)
    @ApiOperation(value = "Get a data set by id",
            notes = "Get a data set content based on provided id. Id should be a UUID returned by the list operation. Not valid or non existing data set id returns empty content.")
    @Timed
    @ResponseBody
    public Callable<DataSet> get(
            @RequestParam(defaultValue = "true") @ApiParam(name = "metadata",
                    value = "Include metadata information in the response") boolean metadata, //
            @RequestParam(defaultValue = "false") @ApiParam(name = "includeInternalContent",
                    value = "Include internal content in the response") boolean includeInternalContent, //
            @RequestParam(defaultValue = "-1") @ApiParam(name = STORAGE_LIMIT, value = STORAGE_LIMIT) long limit, //
            @ApiParam(value = "Filter for retrieved content.") @RequestParam(value = "filter",
                    defaultValue = "") String filter,
            @PathVariable(value = "id") @ApiParam(name = "id",
                    value = "Id of the requested data set") String dataSetId) {
        return () -> {
            final Marker marker = Markers.dataset(dataSetId);
            LOG.debug(marker, "Get data set #{}", dataSetId);
            Stream<DataSetRow> stream = null;
            try {
                DataSetMetadata dataSetMetadata = dataSetMetadataRepository.get(dataSetId);
                assertDataSetMetadata(dataSetMetadata, dataSetId);
                // Build the result
                DataSet dataSet = new DataSet();
                if (metadata) {
                    dataSet.setMetadata(conversionService.convert(dataSetMetadata, UserDataSetMetadata.class));
                }
                stream = contentStore.stream(dataSetMetadata, limit); // Disable line limit

                // on-demand analyzer for dataset (See TDP-4404, migration problems)
                if (dataSetMetadata.getRowMetadata().getColumns().stream().anyMatch(
                        c -> c.getStatistics().getWordPatternFrequencyTable().isEmpty())) {
                    stream = insertWordPatternAnalysis(dataSetMetadata, stream);
                }

                if (!includeInternalContent) {
                    LOG.debug("Skip internal content when serving data set #{} content.", dataSetId);
                    stream = stream.map(r -> {
                        final Map<String, Object> values = r.values();
                        final Map<String, Object> filteredValues = new HashMap<>(values);
                        // Remove technical properties from returned values.
                        values.forEach((k, v) -> {
                            if (k != null && k.startsWith(FlagNames.INTERNAL_PROPERTY_PREFIX)) {
                                filteredValues.remove(k);
                            }
                        });
                        filteredValues.put(FlagNames.TDP_ID, r.getTdpId()); // Include TDP_ID anyway
                        return new DataSetRow(r.getRowMetadata(), filteredValues);
                    });
                }

                // Filter content
                stream = stream.filter(filterService.build(filter, dataSetMetadata.getRowMetadata()));

                dataSet.setRecords(stream);
                return dataSet;
            } catch (Exception e) {
                if (stream != null) {
                    stream.close();
                }
                throw e;
            } finally {
                LOG.debug(marker, "Get done.");
            }
        };
    }

}