perf(collections): split native chunks without per-line async events

This commit is contained in:
zarzet committed 2026-10-07 04:31:52 +07:00
1 parent 11d496effc
commit 0ca60d4a1b
3 files changed
+314 -6

No files matched your search

@@ -449,10 +449,8 @@ Future<void> _readNativeCollectionRows(
var count = 0;
_CollectionRows? previousKind;
var batch = <Map<String, dynamic>>[];
try {
await for (final line in File(
input.path,
).openRead().transform(utf8.decoder).transform(const LineSplitter())) {
final lines = const LineSplitter().startChunkedConversion(
_CollectionLineSink((line) {
final decoded = jsonDecode(line) as Map<String, dynamic>;
final kind = switch (decoded['kind']) {
'wishlist' || 'playlist_track' => _CollectionRows.wishlist,
@@ -466,15 +464,26 @@ Future<void> _readNativeCollectionRows(
};
if (previousKind != null &&
(kind != previousKind || batch.length == 256)) {
accumulator.add(previousKind, batch);
accumulator.add(previousKind!, batch);
batch = <Map<String, dynamic>>[];
}
previousKind = kind;
batch.add(Map<String, dynamic>.from(decoded['row'] as Map));
count++;
}),
);
try {
// Decode file chunks asynchronously, then split and consume their lines
// synchronously on this worker. A stream event per row adds avoidable
// scheduling overhead to large snapshots.
await for (final chunk in File(
input.path,
).openRead().transform(utf8.decoder)) {
lines.add(chunk);
}
lines.close();
if (previousKind != null && batch.isNotEmpty) {
accumulator.add(previousKind, batch);
accumulator.add(previousKind!, batch);
}
if (input.count != null && count != input.count) {
throw const FormatException('Incomplete native collection snapshot');
@@ -484,6 +493,17 @@ Future<void> _readNativeCollectionRows(
}
}
class _CollectionLineSink implements Sink<String> {
const _CollectionLineSink(this.onLine);
final void Function(String) onLine;
@override
void add(String line) => onLine(line);
@override
void close() {}
}
enum _CollectionExportRows { wishlist, loved, artist, playlistTrack }
typedef _CollectionExportBatch = ({
@@ -78,6 +78,164 @@ LibraryCollectionsSnapshot collectionFixture(int count) =>
);
void main() {
Future<LibraryCollectionsState> nativeBytes(
List<int> bytes,
int count,
Directory directory,
) async {
final file = File('${directory.path}/rows.ndjson');
await file.writeAsBytes(bytes);
return hydrateLibraryCollectionsSnapshot(
LibraryCollectionsSnapshot(
wishlistRows: const [],
lovedRows: const [],
playlistRows: const [],
playlistTrackRows: const [],
favoriteArtistRows: const [],
nativeRowsPath: file.path,
nativeRowsDirectory: directory.path,
nativeRowCount: count,
),
);
}
String nativeLine(String kind, Map<String, dynamic> row) =>
jsonEncode({'kind': kind, 'row': row});
test(
'native chunked reader retains interleaved group order across 256-row batches',
() async {
final directory = await Directory.systemTemp.createTemp(
'collection-row-groups-',
);
final wishlist = List.generate(513, trackRow);
final loved = List.generate(3, (index) => trackRow(900 + index));
final snapshot = LibraryCollectionsSnapshot(
wishlistRows: wishlist,
lovedRows: loved,
favoriteArtistRows: const [],
playlistRows: const [],
playlistTrackRows: const [],
);
final lines = [
for (final row in wishlist.take(257)) nativeLine('wishlist', row),
nativeLine('loved', loved[0]),
for (final row in wishlist.skip(257)) nativeLine('wishlist', row),
for (final row in loved.skip(1)) nativeLine('loved', row),
];
// Include LF, CRLF and lone CR; the last row intentionally has no newline.
final text = StringBuffer();
for (var index = 0; index < lines.length; index++) {
text.write(lines[index]);
if (index + 1 != lines.length) {
text.write(['\n', '\r\n', '\r'][index % 3]);
}
}
final actual = await nativeBytes(
utf8.encode(text.toString()),
lines.length,
directory,
);
expect(
jsonEncode(actual.toJson()),
jsonEncode(decodeLibraryCollectionsSnapshot(snapshot).toJson()),
);
expect(
actual.wishlist.map((entry) => entry.key),
wishlist.map((row) => row['track_key']),
);
expect(
actual.loved.map((entry) => entry.key),
loved.map((row) => row['track_key']),
);
expect(await directory.exists(), isFalse);
},
);
test(
'native UTF-8 and CRLF survive exact file chunk boundaries and unterminated tail',
() async {
Map<String, dynamic> commentRow(int id, String comment) => {
...trackRow(id),
'track_json': jsonEncode(
(jsonDecode(trackRow(id)['track_json'] as String)
as Map<String, dynamic>)
..['comment'] = comment,
),
};
final initial = nativeLine('wishlist', commentRow(0, '🎵終わり'));
final emojiPrefix = utf8
.encode(initial.substring(0, initial.indexOf('🎵')))
.length;
final unicodeRow = commentRow(0, '${'x' * (65535 - emojiPrefix)}🎵終わり');
final unicodeLine = nativeLine('wishlist', unicodeRow);
final unicodeBytes = utf8.encode(unicodeLine);
expect(unicodeBytes[65535], 0xf0);
expect(unicodeBytes[65536], 0x9f);
final unicodeDirectory = await Directory.systemTemp.createTemp(
'collection-utf8-boundary-',
);
final unicode = await nativeBytes(unicodeBytes, 1, unicodeDirectory);
expect(
unicode.wishlist.single.track.comment,
(jsonDecode(unicodeRow['track_json'] as String) as Map)['comment'],
);
expect(await unicodeDirectory.exists(), isFalse);
final unpadded = nativeLine('wishlist', commentRow(1, ''));
final crRow = commentRow(1, 'y' * (65535 - utf8.encode(unpadded).length));
final crLine = nativeLine('wishlist', crRow);
final tail = trackRow(2);
final crBytes = utf8.encode('$crLine\r\n${nativeLine('wishlist', tail)}');
expect(crBytes[65535], 13);
expect(crBytes[65536], 10);
final crDirectory = await Directory.systemTemp.createTemp(
'collection-crlf-boundary-',
);
final crlf = await nativeBytes(crBytes, 2, crDirectory);
expect(crlf.wishlist.map((entry) => entry.key), [
'example:1',
'example:2',
]);
expect(
crlf.wishlist.first.track.comment,
'y' * (65535 - utf8.encode(unpadded).length),
);
expect(await crDirectory.exists(), isFalse);
},
);
test(
'native malformed bytes/envelopes and row-count mismatch always remove staging',
() async {
final valid = nativeLine('wishlist', trackRow(1));
final invalid = <({List<int> bytes, int count})>[
(bytes: [...utf8.encode('$valid\n'), 0xc3, 0x28], count: 2),
(bytes: utf8.encode('$valid\n{"kind":'), count: 2),
(bytes: utf8.encode('$valid\n\n'), count: 2),
(bytes: utf8.encode(nativeLine('unknown', {})), count: 1),
(
bytes: utf8.encode(
jsonEncode({'kind': 'wishlist', 'row': <dynamic>[]}),
),
count: 1,
),
(bytes: utf8.encode(valid), count: 0),
(bytes: utf8.encode('$valid\r\n'), count: 2),
];
for (final input in invalid) {
final directory = await Directory.systemTemp.createTemp(
'collection-invalid-chunks-',
);
await expectLater(
nativeBytes(input.bytes, input.count, directory),
throwsA(isA<Error>()),
);
expect(await directory.exists(), isFalse);
}
},
);
test(
'native row file preserves snapshot parity and is always cleaned',
() async {
@@ -0,0 +1,130 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
// Parse-only experiment: both readers perform the same outer/inner JSON
// decoding and ordered checksum. It excludes SQL, native serialization,
// isolate startup/transfers, model hydration and membership construction.
class _DecodedRows implements Sink<String> {
int count = 0;
int checksum = 0;
@override
void add(String line) {
final envelope = jsonDecode(line) as Map<String, dynamic>;
final row = Map<String, dynamic>.from(envelope['row'] as Map);
final track =
jsonDecode(row['track_json'] as String) as Map<String, dynamic>;
final text =
'${envelope['kind']}|${row['track_key']}|${track['name']}|${track['comment']}';
for (final unit in text.codeUnits) {
checksum = ((checksum * 31) + unit) & 0x7fffffff;
}
count++;
}
@override
void close() {}
}
Future<({int count, int checksum})> _parse(File file, bool chunked) async {
final rows = _DecodedRows();
if (chunked) {
final lines = const LineSplitter().startChunkedConversion(rows);
await for (final chunk in file.openRead().transform(utf8.decoder)) {
lines.add(chunk);
}
lines.close();
} else {
await for (final line
in file
.openRead()
.transform(utf8.decoder)
.transform(const LineSplitter())) {
rows.add(line);
}
}
return (count: rows.count, checksum: rows.checksum);
}
void main() {
test('paired native collection rows parse-only benchmark', () async {
final directory = await Directory.systemTemp.createTemp(
'collection-parser-benchmark-',
);
try {
final file = File('${directory.path}/rows.ndjson');
const count = 8000;
await file.writeAsString(
[
for (var index = 0; index < count; index++)
jsonEncode({
'kind': index.isEven ? 'wishlist' : 'loved',
'row': {
'track_key': 'example:$index',
'added_at': '2026-10-07T00:00:00.000Z',
'track_json': jsonEncode({
'id': '$index',
'name': 'Song $index 音楽 🎵',
'artistName': 'Artist ${index % 100}',
'albumName': 'Album ${index ~/ 12}',
'duration': 180,
'comment': '"quoted" \\ / \r\n ${'x' * 512}',
'coverUrl': 'https://example.test/$index.jpg',
'availability': {'tidal': true, 'qobuz': false},
}),
},
}),
].join('\n'),
);
final expected = await _parse(file, false);
expect(expected.count, count);
// Warm both codecs/file paths before alternating paired measurements.
for (var warm = 0; warm < 2; warm++) {
expect(await _parse(file, true), expected);
expect(await _parse(file, false), expected);
}
final old = <int>[];
final chunked = <int>[];
for (var sample = 0; sample < 7; sample++) {
for (final useChunked
in sample.isEven ? [false, true] : [true, false]) {
final watch = Stopwatch()..start();
final actual = await _parse(file, useChunked);
watch.stop();
expect(actual, expected);
(useChunked ? chunked : old).add(watch.elapsedMicroseconds);
}
}
int median(List<int> samples) =>
(List<int>.of(samples)..sort())[samples.length ~/ 2];
final report = {
'scope':
'parse only: file read, strict UTF-8 decode, outer/inner JSON decode, ordered checksum; excludes full collection loading/hydration',
'platform': Platform.operatingSystem,
'mode': 'flutter test host',
'rows': count,
'bytes': await file.length(),
'checksum': expected.checksum,
'async_line_stream_us': old,
'chunked_line_sink_us': chunked,
'async_line_stream_median_us': median(old),
'chunked_line_sink_median_us': median(chunked),
'parity': true,
};
const output = String.fromEnvironment('COLLECTION_ROWS_PARSER_OUTPUT');
if (output.isNotEmpty) {
final result = File(output);
await result.parent.create(recursive: true);
await result.writeAsString(
const JsonEncoder.withIndent(' ').convert(report),
);
}
// ignore: avoid_print
print('COLLECTION_ROWS_PARSER_REPORT ${jsonEncode(report)}');
} finally {
await directory.delete(recursive: true);
}
}, skip: !const bool.fromEnvironment('RUN_COLLECTION_ROWS_PARSER_BENCHMARK'));
}