mirror of
https://github.com/apache/arrow.git
synced 2026-10-29 00:56:31 +08:00
incorporate first suggestions for writing EOS in File
This commit is contained in:
1 parent
901dfeab02
commit
9e83053d0b
3 files changed
+7
-5
No files matched your search
@@ -999,6 +999,9 @@ class StreamBookKeeper {
|
||||
int64_t position_;
|
||||
};
|
||||
|
||||
// End of stream marker
|
||||
constexpr int32_t kEos = 0;
|
||||
|
||||
/// A IpcPayloadWriter implementation that writes to a IPC stream
|
||||
/// (with an end-of-stream marker)
|
||||
class PayloadStreamWriter : public internal::IpcPayloadWriter,
|
||||
@@ -1022,7 +1025,6 @@ class PayloadStreamWriter : public internal::IpcPayloadWriter,
|
||||
|
||||
Status Close() override {
|
||||
// Write 0 EOS message
|
||||
const int32_t kEos = 0;
|
||||
return Write(&kEos, sizeof(int32_t));
|
||||
}
|
||||
};
|
||||
@@ -1076,10 +1078,8 @@ class PayloadFileWriter : public internal::IpcPayloadWriter, protected StreamBoo
|
||||
}
|
||||
|
||||
Status Close() override {
|
||||
// Close stream for compatibility with sequential readers
|
||||
RETURN_NOT_OK(UpdatePosition());
|
||||
int32_t end_of_stream_marker = 0;
|
||||
RETURN_NOT_OK(Write(&end_of_stream_marker, sizeof(int32_t)));
|
||||
// Write 0 EOS message for compatibility with sequential readers
|
||||
RETURN_NOT_OK(Write(&kEos, sizeof(int32_t)));
|
||||
|
||||
// Write file footer
|
||||
RETURN_NOT_OK(UpdatePosition());
|
||||
|
||||
@@ -65,6 +65,7 @@ public class ArrowFileWriter extends ArrowWriter {
|
||||
|
||||
@Override
|
||||
protected void endInternal(WriteChannel out) throws IOException {
|
||||
out.writeIntLittleEndian(0); // Write an EOS identifier for sequential readers
|
||||
long footerStart = out.getCurrentPosition();
|
||||
out.write(new ArrowFooter(schema, dictionaryBlocks, recordBlocks), false);
|
||||
int footerLength = (int) (out.getCurrentPosition() - footerStart);
|
||||
|
||||
@@ -314,6 +314,7 @@ export class RecordBatchFileWriter<T extends { [key: string]: DataType } = any>
|
||||
this._recordBatchBlocks, this._dictionaryBlocks
|
||||
));
|
||||
return this
|
||||
._writePadding(4) // eos bytes for sequential readers
|
||||
._write(buffer) // Write the flatbuffer
|
||||
._write(Int32Array.of(buffer.byteLength)) // then the footer size suffix
|
||||
._writeMagic(); // then the magic suffix
|
||||
|
||||
Reference in new issue
Block a user