@@ -61,6 +61,19 @@ public ArrowFileWriter(Stream stream, Schema schema, bool leaveOpen, IpcOptions
6161 RecordBatchBlocks = new List < Block > ( ) ;
6262 }
6363
64+ public override void WriteRecordBatch ( RecordBatch recordBatch )
65+ {
66+ // TODO: Compare record batch schema
67+
68+ if ( ! HasWrittenHeader )
69+ {
70+ WriteHeader ( ) ;
71+ HasWrittenHeader = true ;
72+ }
73+
74+ WriteRecordBatchInternal ( recordBatch ) ;
75+ }
76+
6477 public override async Task WriteRecordBatchAsync ( RecordBatch recordBatch , CancellationToken cancellationToken = default )
6578 {
6679 // TODO: Compare record batch schema
@@ -104,13 +117,28 @@ private protected override void FinishedWritingRecordBatch(long bodyLength, long
104117 _currentRecordBatchOffset = - 1 ;
105118 }
106119
120+ private protected override void WriteEndInternal ( )
121+ {
122+ base . WriteEndInternal ( ) ;
123+
124+ WriteFooter ( Schema ) ;
125+ }
126+
107127 private protected override async ValueTask WriteEndInternalAsync ( CancellationToken cancellationToken )
108128 {
109129 await base . WriteEndInternalAsync ( cancellationToken ) ;
110130
111131 await WriteFooterAsync ( Schema , cancellationToken ) ;
112132 }
113133
134+ private void WriteHeader ( )
135+ {
136+ // Write magic number and empty padding up to the 8-byte boundary
137+
138+ WriteMagic ( ) ;
139+ WritePadding ( CalculatePadding ( ArrowFileConstants . Magic . Length ) ) ;
140+ }
141+
114142 private async Task WriteHeaderAsync ( CancellationToken cancellationToken )
115143 {
116144 // Write magic number and empty padding up to the 8-byte boundary
@@ -120,6 +148,64 @@ await WritePaddingAsync(CalculatePadding(ArrowFileConstants.Magic.Length))
120148 . ConfigureAwait ( false ) ;
121149 }
122150
151+ private void WriteFooter ( Schema schema )
152+ {
153+ Builder . Clear ( ) ;
154+
155+ long offset = BaseStream . Position ;
156+
157+ // Serialize the schema
158+
159+ FlatBuffers . Offset < Flatbuf . Schema > schemaOffset = SerializeSchema ( schema ) ;
160+
161+ // Serialize all record batches
162+
163+ Flatbuf . Footer . StartRecordBatchesVector ( Builder , RecordBatchBlocks . Count ) ;
164+
165+ foreach ( Block recordBatch in RecordBatchBlocks )
166+ {
167+ Flatbuf . Block . CreateBlock (
168+ Builder , recordBatch . Offset , recordBatch . MetadataLength , recordBatch . BodyLength ) ;
169+ }
170+
171+ FlatBuffers . VectorOffset recordBatchesVectorOffset = Builder . EndVector ( ) ;
172+
173+ // Serialize all dictionaries
174+ // NOTE: Currently unsupported.
175+
176+ Flatbuf . Footer . StartDictionariesVector ( Builder , 0 ) ;
177+
178+ FlatBuffers . VectorOffset dictionaryBatchesOffset = Builder . EndVector ( ) ;
179+
180+ // Serialize and write the footer flatbuffer
181+
182+ FlatBuffers . Offset < Flatbuf . Footer > footerOffset = Flatbuf . Footer . CreateFooter ( Builder , CurrentMetadataVersion ,
183+ schemaOffset , dictionaryBatchesOffset , recordBatchesVectorOffset ) ;
184+
185+ Builder . Finish ( footerOffset . Value ) ;
186+
187+ WriteFlatBuffer ( ) ;
188+
189+ // Write footer length
190+
191+ Buffers . RentReturn ( 4 , ( buffer ) =>
192+ {
193+ int footerLength ;
194+ checked
195+ {
196+ footerLength = ( int ) ( BaseStream . Position - offset ) ;
197+ }
198+
199+ BinaryPrimitives . WriteInt32LittleEndian ( buffer . Span , footerLength ) ;
200+
201+ BaseStream . Write ( buffer ) ;
202+ } ) ;
203+
204+ // Write magic
205+
206+ WriteMagic ( ) ;
207+ }
208+
123209 private async Task WriteFooterAsync ( Schema schema , CancellationToken cancellationToken )
124210 {
125211 Builder . Clear ( ) ;
@@ -182,6 +268,11 @@ await Buffers.RentReturnAsync(4, async (buffer) =>
182268 await WriteMagicAsync ( cancellationToken ) . ConfigureAwait ( false ) ;
183269 }
184270
271+ private void WriteMagic ( )
272+ {
273+ BaseStream . Write ( ArrowFileConstants . Magic ) ;
274+ }
275+
185276 private ValueTask WriteMagicAsync ( CancellationToken cancellationToken )
186277 {
187278 return BaseStream . WriteAsync ( ArrowFileConstants . Magic , cancellationToken ) ;
0 commit comments