public static List<DataBlockIndex> buildDataBlockIndicesFromGroup(
        List<List<StreamDataBlock>> streamDataBlockGroup) {
        List<DataBlockIndex> dataBlockIndices = new ArrayList<>();

        long blockStartPosition = 0;
        for (List<StreamDataBlock> streamDataBlocks : streamDataBlockGroup) {
            if (streamDataBlocks.isEmpty()) {
                continue;
            }
            dataBlockIndices.add(new DataBlockIndex(
                streamDataBlocks.get(0).getStreamId(),
                streamDataBlocks.get(0).getStartOffset(),
                (int) (streamDataBlocks.get(streamDataBlocks.size() - 1).getEndOffset() - streamDataBlocks.get(0).getStartOffset()),
                streamDataBlocks.stream().map(StreamDataBlock::dataBlockIndex).mapToInt(DataBlockIndex::recordCount).sum(),
                blockStartPosition,
                streamDataBlocks.stream().mapToInt(StreamDataBlock::getBlockSize).sum()));
            blockStartPosition += streamDataBlocks.stream().mapToInt(StreamDataBlock::getBlockSize).sum();
        }

        return dataBlockIndices;
    }
Group stream data blocks by certain conditions.

@param streamDataBlocks stream data blocks to be grouped
@param predicate        the predicate to check whether a stream data block should be grouped with the previous one
@return grouped stream data blocks
s3stream/src/main/java/com/automq/stream/s3/compact/utils/CompactionUtils.java