@Test public void testTimeFormatSDF() throws Exception { FileUtils.deleteQuietly(WORKING_DIR); RealtimeToOfflineSegmentsTaskExecutor realtimeToOfflineSegmentsTaskExecutor = new RealtimeToOfflineSegmentsTaskExecutor(); Map<String, String> configs = new HashMap<>(); configs.put(MinionConstants.TABLE_NAME_KEY, TABLE_NAME_SDF); configs.put(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY, "1600473600000"); configs.put(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY, "1600560000000"); configs.put(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY, "rollup"); PinotTaskConfig pinotTaskConfig = new PinotTaskConfig(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE, configs); List<SegmentConversionResult> conversionResults = realtimeToOfflineSegmentsTaskExecutor.convert(pinotTaskConfig, _segmentIndexDirListSDF, WORKING_DIR); assertEquals(conversionResults.size(), 1); File resultingSegment = conversionResults.get(0).getFile(); SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(resultingSegment); assertEquals(segmentMetadata.getTotalDocs(), 3); ColumnMetadata columnMetadataForT = segmentMetadata.getColumnMetadataFor(T_TRX); assertEquals(columnMetadataForT.getCardinality(), 3); assertTrue((int) columnMetadataForT.getMinValue() >= 2020091900); assertTrue((int) columnMetadataForT.getMaxValue() < 2020092000); }
@Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { String taskType = pinotTaskConfig.getTaskType(); Map<String, String> configs = pinotTaskConfig.getConfigs(); LOGGER.info("Starting task: {} with configs: {}", taskType, configs); long startMillis = System.currentTimeMillis(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); Schema schema = getSchema(tableNameWithType); Set<String> schemaColumns = schema.getPhysicalColumnNames(); String timeColumn = tableConfig.getValidationConfig().getTimeColumnName(); DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn); Preconditions .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s", timeColumn, tableNameWithType); long windowStartMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY)); long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY)); String timeColumnTransformFunction = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY); String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY); Map<String, String> aggregatorConfigs = new HashMap<>(); for (Map.Entry<String, String> entry : configs.entrySet()) { String key = entry.getKey(); if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) { String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0]; aggregatorConfigs.put(column, entry.getValue()); } } String numRecordsPerSegment = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY); SegmentProcessorConfig.Builder segmentProcessorConfigBuilder = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema); if (timeColumnTransformFunction != null) { RecordTransformerConfig recordTransformerConfig = getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn); segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig); } RecordFilterConfig recordFilterConfig = getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn); segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig); if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) { Map<String, ColumnPartitionConfig> columnPartitionMap = tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap(); PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns); segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig)); } List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn(); CollectorConfig collectorConfig = getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns); segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig); if (numRecordsPerSegment != null) { SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment); segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig); } SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build(); File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR); Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s", inputSegmentsDir.getAbsolutePath(), taskType); for (File indexDir : originalIndexDirs) { FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir); } File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR); Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s", outputSegmentsDir.getAbsolutePath(), taskType); SegmentProcessorFramework segmentProcessorFramework = new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir); try { segmentProcessorFramework.processSegments(); } finally { segmentProcessorFramework.cleanup(); } long endMillis = System.currentTimeMillis(); LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis)); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : outputSegmentsDir.listFiles()) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }
RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { String taskType = pinotTaskConfig.getTaskType(); Map<String, String> configs = pinotTaskConfig.getConfigs(); LOGGER.info("Starting task: {} with configs: {}", taskType, configs); long startMillis = System.currentTimeMillis(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); Schema schema = getSchema(tableNameWithType); Set<String> schemaColumns = schema.getPhysicalColumnNames(); String timeColumn = tableConfig.getValidationConfig().getTimeColumnName(); DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn); Preconditions .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s", timeColumn, tableNameWithType); long windowStartMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY)); long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY)); String timeColumnTransformFunction = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY); String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY); Map<String, String> aggregatorConfigs = new HashMap<>(); for (Map.Entry<String, String> entry : configs.entrySet()) { String key = entry.getKey(); if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) { String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0]; aggregatorConfigs.put(column, entry.getValue()); } } String numRecordsPerSegment = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY); SegmentProcessorConfig.Builder segmentProcessorConfigBuilder = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema); if (timeColumnTransformFunction != null) { RecordTransformerConfig recordTransformerConfig = getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn); segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig); } RecordFilterConfig recordFilterConfig = getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn); segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig); if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) { Map<String, ColumnPartitionConfig> columnPartitionMap = tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap(); PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns); segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig)); } List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn(); CollectorConfig collectorConfig = getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns); segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig); if (numRecordsPerSegment != null) { SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment); segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig); } SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build(); File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR); Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s", inputSegmentsDir.getAbsolutePath(), taskType); for (File indexDir : originalIndexDirs) { FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir); } File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR); Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s", outputSegmentsDir.getAbsolutePath(), taskType); SegmentProcessorFramework segmentProcessorFramework = new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir); try { segmentProcessorFramework.processSegments(); } finally { segmentProcessorFramework.cleanup(); } long endMillis = System.currentTimeMillis(); LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis)); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : outputSegmentsDir.listFiles()) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; } }
RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { String taskType = pinotTaskConfig.getTaskType(); Map<String, String> configs = pinotTaskConfig.getConfigs(); LOGGER.info("Starting task: {} with configs: {}", taskType, configs); long startMillis = System.currentTimeMillis(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); Schema schema = getSchema(tableNameWithType); Set<String> schemaColumns = schema.getPhysicalColumnNames(); String timeColumn = tableConfig.getValidationConfig().getTimeColumnName(); DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn); Preconditions .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s", timeColumn, tableNameWithType); long windowStartMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY)); long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY)); String timeColumnTransformFunction = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY); String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY); Map<String, String> aggregatorConfigs = new HashMap<>(); for (Map.Entry<String, String> entry : configs.entrySet()) { String key = entry.getKey(); if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) { String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0]; aggregatorConfigs.put(column, entry.getValue()); } } String numRecordsPerSegment = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY); SegmentProcessorConfig.Builder segmentProcessorConfigBuilder = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema); if (timeColumnTransformFunction != null) { RecordTransformerConfig recordTransformerConfig = getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn); segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig); } RecordFilterConfig recordFilterConfig = getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn); segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig); if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) { Map<String, ColumnPartitionConfig> columnPartitionMap = tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap(); PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns); segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig)); } List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn(); CollectorConfig collectorConfig = getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns); segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig); if (numRecordsPerSegment != null) { SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment); segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig); } SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build(); File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR); Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s", inputSegmentsDir.getAbsolutePath(), taskType); for (File indexDir : originalIndexDirs) { FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir); } File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR); Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s", outputSegmentsDir.getAbsolutePath(), taskType); SegmentProcessorFramework segmentProcessorFramework = new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir); try { segmentProcessorFramework.processSegments(); } finally { segmentProcessorFramework.cleanup(); } long endMillis = System.currentTimeMillis(); LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis)); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : outputSegmentsDir.listFiles()) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }  }
RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { String taskType = pinotTaskConfig.getTaskType(); Map<String, String> configs = pinotTaskConfig.getConfigs(); LOGGER.info("Starting task: {} with configs: {}", taskType, configs); long startMillis = System.currentTimeMillis(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); Schema schema = getSchema(tableNameWithType); Set<String> schemaColumns = schema.getPhysicalColumnNames(); String timeColumn = tableConfig.getValidationConfig().getTimeColumnName(); DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn); Preconditions .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s", timeColumn, tableNameWithType); long windowStartMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY)); long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY)); String timeColumnTransformFunction = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY); String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY); Map<String, String> aggregatorConfigs = new HashMap<>(); for (Map.Entry<String, String> entry : configs.entrySet()) { String key = entry.getKey(); if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) { String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0]; aggregatorConfigs.put(column, entry.getValue()); } } String numRecordsPerSegment = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY); SegmentProcessorConfig.Builder segmentProcessorConfigBuilder = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema); if (timeColumnTransformFunction != null) { RecordTransformerConfig recordTransformerConfig = getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn); segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig); } RecordFilterConfig recordFilterConfig = getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn); segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig); if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) { Map<String, ColumnPartitionConfig> columnPartitionMap = tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap(); PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns); segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig)); } List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn(); CollectorConfig collectorConfig = getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns); segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig); if (numRecordsPerSegment != null) { SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment); segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig); } SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build(); File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR); Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s", inputSegmentsDir.getAbsolutePath(), taskType); for (File indexDir : originalIndexDirs) { FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir); } File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR); Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s", outputSegmentsDir.getAbsolutePath(), taskType); SegmentProcessorFramework segmentProcessorFramework = new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir); try { segmentProcessorFramework.processSegments(); } finally { segmentProcessorFramework.cleanup(); } long endMillis = System.currentTimeMillis(); LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis)); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : outputSegmentsDir.listFiles()) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }   }
RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { String taskType = pinotTaskConfig.getTaskType(); Map<String, String> configs = pinotTaskConfig.getConfigs(); LOGGER.info("Starting task: {} with configs: {}", taskType, configs); long startMillis = System.currentTimeMillis(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); Schema schema = getSchema(tableNameWithType); Set<String> schemaColumns = schema.getPhysicalColumnNames(); String timeColumn = tableConfig.getValidationConfig().getTimeColumnName(); DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn); Preconditions .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s", timeColumn, tableNameWithType); long windowStartMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY)); long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY)); String timeColumnTransformFunction = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY); String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY); Map<String, String> aggregatorConfigs = new HashMap<>(); for (Map.Entry<String, String> entry : configs.entrySet()) { String key = entry.getKey(); if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) { String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0]; aggregatorConfigs.put(column, entry.getValue()); } } String numRecordsPerSegment = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY); SegmentProcessorConfig.Builder segmentProcessorConfigBuilder = new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema); if (timeColumnTransformFunction != null) { RecordTransformerConfig recordTransformerConfig = getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn); segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig); } RecordFilterConfig recordFilterConfig = getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn); segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig); if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) { Map<String, ColumnPartitionConfig> columnPartitionMap = tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap(); PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns); segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig)); } List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn(); CollectorConfig collectorConfig = getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns); segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig); if (numRecordsPerSegment != null) { SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment); segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig); } SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build(); File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR); Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s", inputSegmentsDir.getAbsolutePath(), taskType); for (File indexDir : originalIndexDirs) { FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir); } File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR); Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s", outputSegmentsDir.getAbsolutePath(), taskType); SegmentProcessorFramework segmentProcessorFramework = new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir); try { segmentProcessorFramework.processSegments(); } finally { segmentProcessorFramework.cleanup(); } long endMillis = System.currentTimeMillis(); LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis)); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : outputSegmentsDir.listFiles()) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }    }