diff --git a/src/ipc/writer.ts b/src/ipc/writer.ts index 6554498c..e5246419 100644 --- a/src/ipc/writer.ts +++ b/src/ipc/writer.ts @@ -439,9 +439,11 @@ export class RecordBatchFileWriter extends RecordBatchW this._writeLegacyIpcFormat = false; } - // @ts-ignore protected _writeSchema(schema: Schema) { - return this._writeMagic()._writePadding(2); + // The file format embeds a complete IPC stream after the leading + // magic and padding, so the Schema message has to be written here too. + this._writeMagic()._writePadding(2); + return super._writeSchema(schema); } protected _writeDictionaryBatch(dictionary: Data, id: number, isDelta = false) { diff --git a/test/unit/ipc/writer/file-writer-tests.ts b/test/unit/ipc/writer/file-writer-tests.ts index 3cefa7a8..9c68ff3b 100644 --- a/test/unit/ipc/writer/file-writer-tests.ts +++ b/test/unit/ipc/writer/file-writer-tests.ts @@ -31,6 +31,7 @@ import { RecordBatch, RecordBatchFileWriter, RecordBatchReader, + RecordBatchStreamReader, Table, tableFromArrays, Uint32, @@ -134,6 +135,23 @@ describe('RecordBatchFileWriter', () => { expect(dictionary).toBeInstanceOf(Vector); expect(dictionary.data).toHaveLength(20); }); + + describe('embedded IPC stream', () => { + const cases: [string, Table][] = [ + ['all types', generate.table([10, 20, 30]).table], + ['dictionaries', generateDictionaryTables([10, 20, 30]).next().value!], + ]; + for (const [name, source] of cases) { + it(`can be read by a stream reader after the leading magic (${name})`, async () => { + const bytes = await RecordBatchFileWriter.writeAll(source).toUint8Array(); + // Skip the leading "ARROW1" magic and its two bytes of padding. + const reader = RecordBatchStreamReader.from(bytes.subarray(FILE_FORMAT_HEADER_LENGTH)); + const result = new Table(reader); + expect(result.schema.fields).toHaveLength(source.schema.fields.length); + expect(result).toEqualTable(source); + }); + } + }); }); function testFileWriter(table: Table, name: string, options?: RecordBatchStreamWriterOptions) {