@Test public void testSegmentRollupWithTimeConversion() throws Exception { SegmentConverter segmentConverter = new SegmentConverter.Builder().setTableName(TABLE_NAME).setSegmentName("segmentRollupWithTimeConversion") .setInputIndexDirs(_segmentIndexDirList).setWorkingDir(WORKING_DIR).setSkipTimeValueCheck(false) .setRecordTransformer((row) -> { row.putValue(T, BASE_TIMESTAMP); return row; }).setGroupByColumns(Arrays.asList(D1, D2, T)).setRecordAggregator((rows) -> { GenericRow result = rows.get(0); for (int i = 1; i < rows.size(); i++) { GenericRow current = rows.get(i); Object aggregatedValue = ((Number) result.getValue(M1)).intValue() + ((Number) current.getValue(M1)).intValue(); result.putValue(M1, aggregatedValue); } return result; }).setTotalNumPartition(1).setTableConfig(_tableConfig).build(); List<File> result = segmentConverter.convertSegment(); assertEquals(result.size(), 1); List<GenericRow> outputRows = new ArrayList<>(); try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(result.get(0))) { while (pinotSegmentRecordReader.hasNext()) { outputRows.add(pinotSegmentRecordReader.next()); } } assertEquals(outputRows.size(), NUM_ROWS / REPEAT_ROWS, "Number of rows returned by segment converter is incorrect"); int expectedValue = 0; for (GenericRow row : outputRows) { assertEquals(row.getValue(D1), expectedValue); assertEquals(row.getValue(D2), Integer.toString(expectedValue)); assertEquals(row.getValue(M1), expectedValue * NUM_SEGMENTS * REPEAT_ROWS); assertEquals(row.getValue(T), BASE_TIMESTAMP); expectedValue++; } }
public List<File> convertSegment() throws Exception { List<File> resultFiles = new ArrayList<>(); for (int currentPartition = 0; currentPartition < _totalNumPartition; currentPartition++) { Preconditions.checkNotNull(_recordTransformer); String mapperOutputPath = _workingDir.getPath() + File.separator + MAPPER_PREFIX + currentPartition; String outputSegmentName = (_totalNumPartition <= 1) ? _segmentName : _segmentName + "_" + currentPartition; try (MapperRecordReader mapperRecordReader = new MapperRecordReader(_inputIndexDirs, _recordTransformer, _recordPartitioner, _totalNumPartition, currentPartition)) { buildSegment(mapperOutputPath, outputSegmentName, mapperRecordReader, mapperRecordReader.getSchema(), _tableConfig); } File outputSegment = new File(mapperOutputPath + File.separator + outputSegmentName); if (_recordAggregator != null && _groupByColumns != null && _groupByColumns.size() > 0) { String reducerOutputPath = _workingDir.getPath() + File.separator + REDUCER_PREFIX + currentPartition; try (ReducerRecordReader reducerRecordReader = new ReducerRecordReader(outputSegment, _recordAggregator, _groupByColumns)) { buildSegment(reducerOutputPath, outputSegmentName, reducerRecordReader, reducerRecordReader.getSchema(), _tableConfig); } outputSegment = new File(reducerOutputPath + File.separator + outputSegmentName); } IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); List<String> sortedColumn = indexingConfig.getSortedColumn(); List<String> invertedIndexColumns = indexingConfig.getInvertedIndexColumns(); if (CollectionUtils.isNotEmpty(sortedColumn) || CollectionUtils.isNotEmpty(invertedIndexColumns)) { String indexGenerationOutputPath = _workingDir.getPath() + File.separator + INDEX_PREFIX + currentPartition; try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(outputSegment, null, sortedColumn)) { buildSegment(indexGenerationOutputPath, outputSegmentName, pinotSegmentRecordReader, pinotSegmentRecordReader.getSchema(), _tableConfig); } outputSegment = new File(indexGenerationOutputPath + File.separator + outputSegmentName); } resultFiles.add(outputSegment); } return resultFiles; }
SegmentConverter { public List<File> convertSegment() throws Exception { List<File> resultFiles = new ArrayList<>(); for (int currentPartition = 0; currentPartition < _totalNumPartition; currentPartition++) { Preconditions.checkNotNull(_recordTransformer); String mapperOutputPath = _workingDir.getPath() + File.separator + MAPPER_PREFIX + currentPartition; String outputSegmentName = (_totalNumPartition <= 1) ? _segmentName : _segmentName + "_" + currentPartition; try (MapperRecordReader mapperRecordReader = new MapperRecordReader(_inputIndexDirs, _recordTransformer, _recordPartitioner, _totalNumPartition, currentPartition)) { buildSegment(mapperOutputPath, outputSegmentName, mapperRecordReader, mapperRecordReader.getSchema(), _tableConfig); } File outputSegment = new File(mapperOutputPath + File.separator + outputSegmentName); if (_recordAggregator != null && _groupByColumns != null && _groupByColumns.size() > 0) { String reducerOutputPath = _workingDir.getPath() + File.separator + REDUCER_PREFIX + currentPartition; try (ReducerRecordReader reducerRecordReader = new ReducerRecordReader(outputSegment, _recordAggregator, _groupByColumns)) { buildSegment(reducerOutputPath, outputSegmentName, reducerRecordReader, reducerRecordReader.getSchema(), _tableConfig); } outputSegment = new File(reducerOutputPath + File.separator + outputSegmentName); } IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); List<String> sortedColumn = indexingConfig.getSortedColumn(); List<String> invertedIndexColumns = indexingConfig.getInvertedIndexColumns(); if (CollectionUtils.isNotEmpty(sortedColumn) || CollectionUtils.isNotEmpty(invertedIndexColumns)) { String indexGenerationOutputPath = _workingDir.getPath() + File.separator + INDEX_PREFIX + currentPartition; try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(outputSegment, null, sortedColumn)) { buildSegment(indexGenerationOutputPath, outputSegmentName, pinotSegmentRecordReader, pinotSegmentRecordReader.getSchema(), _tableConfig); } outputSegment = new File(indexGenerationOutputPath + File.separator + outputSegmentName); } resultFiles.add(outputSegment); } return resultFiles; } }
SegmentConverter { public List<File> convertSegment() throws Exception { List<File> resultFiles = new ArrayList<>(); for (int currentPartition = 0; currentPartition < _totalNumPartition; currentPartition++) { Preconditions.checkNotNull(_recordTransformer); String mapperOutputPath = _workingDir.getPath() + File.separator + MAPPER_PREFIX + currentPartition; String outputSegmentName = (_totalNumPartition <= 1) ? _segmentName : _segmentName + "_" + currentPartition; try (MapperRecordReader mapperRecordReader = new MapperRecordReader(_inputIndexDirs, _recordTransformer, _recordPartitioner, _totalNumPartition, currentPartition)) { buildSegment(mapperOutputPath, outputSegmentName, mapperRecordReader, mapperRecordReader.getSchema(), _tableConfig); } File outputSegment = new File(mapperOutputPath + File.separator + outputSegmentName); if (_recordAggregator != null && _groupByColumns != null && _groupByColumns.size() > 0) { String reducerOutputPath = _workingDir.getPath() + File.separator + REDUCER_PREFIX + currentPartition; try (ReducerRecordReader reducerRecordReader = new ReducerRecordReader(outputSegment, _recordAggregator, _groupByColumns)) { buildSegment(reducerOutputPath, outputSegmentName, reducerRecordReader, reducerRecordReader.getSchema(), _tableConfig); } outputSegment = new File(reducerOutputPath + File.separator + outputSegmentName); } IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); List<String> sortedColumn = indexingConfig.getSortedColumn(); List<String> invertedIndexColumns = indexingConfig.getInvertedIndexColumns(); if (CollectionUtils.isNotEmpty(sortedColumn) || CollectionUtils.isNotEmpty(invertedIndexColumns)) { String indexGenerationOutputPath = _workingDir.getPath() + File.separator + INDEX_PREFIX + currentPartition; try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(outputSegment, null, sortedColumn)) { buildSegment(indexGenerationOutputPath, outputSegmentName, pinotSegmentRecordReader, pinotSegmentRecordReader.getSchema(), _tableConfig); } outputSegment = new File(indexGenerationOutputPath + File.separator + outputSegmentName); } resultFiles.add(outputSegment); } return resultFiles; } SegmentConverter(List<File> inputIndexDirs, File workingDir, String tableName, String segmentName,
      int totalNumPartition, RecordTransformer recordTransformer, @Nullable RecordPartitioner recordPartitioner,
      @Nullable RecordAggregator recordAggregator, @Nullable List<String> groupByColumns, TableConfig tableConfig,
      boolean skipTimeValueCheck); }
SegmentConverter { public List<File> convertSegment() throws Exception { List<File> resultFiles = new ArrayList<>(); for (int currentPartition = 0; currentPartition < _totalNumPartition; currentPartition++) { Preconditions.checkNotNull(_recordTransformer); String mapperOutputPath = _workingDir.getPath() + File.separator + MAPPER_PREFIX + currentPartition; String outputSegmentName = (_totalNumPartition <= 1) ? _segmentName : _segmentName + "_" + currentPartition; try (MapperRecordReader mapperRecordReader = new MapperRecordReader(_inputIndexDirs, _recordTransformer, _recordPartitioner, _totalNumPartition, currentPartition)) { buildSegment(mapperOutputPath, outputSegmentName, mapperRecordReader, mapperRecordReader.getSchema(), _tableConfig); } File outputSegment = new File(mapperOutputPath + File.separator + outputSegmentName); if (_recordAggregator != null && _groupByColumns != null && _groupByColumns.size() > 0) { String reducerOutputPath = _workingDir.getPath() + File.separator + REDUCER_PREFIX + currentPartition; try (ReducerRecordReader reducerRecordReader = new ReducerRecordReader(outputSegment, _recordAggregator, _groupByColumns)) { buildSegment(reducerOutputPath, outputSegmentName, reducerRecordReader, reducerRecordReader.getSchema(), _tableConfig); } outputSegment = new File(reducerOutputPath + File.separator + outputSegmentName); } IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); List<String> sortedColumn = indexingConfig.getSortedColumn(); List<String> invertedIndexColumns = indexingConfig.getInvertedIndexColumns(); if (CollectionUtils.isNotEmpty(sortedColumn) || CollectionUtils.isNotEmpty(invertedIndexColumns)) { String indexGenerationOutputPath = _workingDir.getPath() + File.separator + INDEX_PREFIX + currentPartition; try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(outputSegment, null, sortedColumn)) { buildSegment(indexGenerationOutputPath, outputSegmentName, pinotSegmentRecordReader, pinotSegmentRecordReader.getSchema(), _tableConfig); } outputSegment = new File(indexGenerationOutputPath + File.separator + outputSegmentName); } resultFiles.add(outputSegment); } return resultFiles; } SegmentConverter(List<File> inputIndexDirs, File workingDir, String tableName, String segmentName,
      int totalNumPartition, RecordTransformer recordTransformer, @Nullable RecordPartitioner recordPartitioner,
      @Nullable RecordAggregator recordAggregator, @Nullable List<String> groupByColumns, TableConfig tableConfig,
      boolean skipTimeValueCheck); List<File> convertSegment(); }
SegmentConverter { public List<File> convertSegment() throws Exception { List<File> resultFiles = new ArrayList<>(); for (int currentPartition = 0; currentPartition < _totalNumPartition; currentPartition++) { Preconditions.checkNotNull(_recordTransformer); String mapperOutputPath = _workingDir.getPath() + File.separator + MAPPER_PREFIX + currentPartition; String outputSegmentName = (_totalNumPartition <= 1) ? _segmentName : _segmentName + "_" + currentPartition; try (MapperRecordReader mapperRecordReader = new MapperRecordReader(_inputIndexDirs, _recordTransformer, _recordPartitioner, _totalNumPartition, currentPartition)) { buildSegment(mapperOutputPath, outputSegmentName, mapperRecordReader, mapperRecordReader.getSchema(), _tableConfig); } File outputSegment = new File(mapperOutputPath + File.separator + outputSegmentName); if (_recordAggregator != null && _groupByColumns != null && _groupByColumns.size() > 0) { String reducerOutputPath = _workingDir.getPath() + File.separator + REDUCER_PREFIX + currentPartition; try (ReducerRecordReader reducerRecordReader = new ReducerRecordReader(outputSegment, _recordAggregator, _groupByColumns)) { buildSegment(reducerOutputPath, outputSegmentName, reducerRecordReader, reducerRecordReader.getSchema(), _tableConfig); } outputSegment = new File(reducerOutputPath + File.separator + outputSegmentName); } IndexingConfig indexingConfig = _tableConfig.getIndexingConfig(); List<String> sortedColumn = indexingConfig.getSortedColumn(); List<String> invertedIndexColumns = indexingConfig.getInvertedIndexColumns(); if (CollectionUtils.isNotEmpty(sortedColumn) || CollectionUtils.isNotEmpty(invertedIndexColumns)) { String indexGenerationOutputPath = _workingDir.getPath() + File.separator + INDEX_PREFIX + currentPartition; try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(outputSegment, null, sortedColumn)) { buildSegment(indexGenerationOutputPath, outputSegmentName, pinotSegmentRecordReader, pinotSegmentRecordReader.getSchema(), _tableConfig); } outputSegment = new File(indexGenerationOutputPath + File.separator + outputSegmentName); } resultFiles.add(outputSegment); } return resultFiles; } SegmentConverter(List<File> inputIndexDirs, File workingDir, String tableName, String segmentName,
      int totalNumPartition, RecordTransformer recordTransformer, @Nullable RecordPartitioner recordPartitioner,
      @Nullable RecordAggregator recordAggregator, @Nullable List<String> groupByColumns, TableConfig tableConfig,
      boolean skipTimeValueCheck); List<File> convertSegment();  }