You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by pa...@apache.org on 2024/10/07 16:02:39 UTC
(arrow-nanoarrow) branch main updated: chore(ci): Ensure integration workflow passes (#643)
This is an automated email from the ASF dual-hosted git repository.
paleolimbot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-nanoarrow.git
The following commit(s) were added to refs/heads/main by this push:
new 19d35d71 chore(ci): Ensure integration workflow passes (#643)
19d35d71 is described below
commit 19d35d718e42e49bb469ca53042718ff5eb04bc9
Author: Dewey Dunnington <de...@dunnington.ca>
AuthorDate: Mon Oct 7 11:02:34 2024 -0500
chore(ci): Ensure integration workflow passes (#643)
Closes #641.
Unfortunately we just have to skip checking Rust compatibility due to
https://github.com/apache/arrow-rs/issues/5052 (e.g.,
https://github.com/apache/arrow-rs/pull/6449 ).
This PR also ensures compatibility with big endian Arrow files and Arrow
files from before the continuation token. Support for those had already
been added in the decoder but hadn't made it to the stream reader yet.
Local check:
```bash
# Assumes arrow-testing, arrow-nanoarrow, and arrow are all checked out in the same dir
export gold_dir=../arrow-testing/data/arrow-ipc-stream/integration
export ARROW_NANOARROW_PATH=$(pwd)/build
pip install -e "../arrow/dev/archery/[all]"
archery integration --with-nanoarrow=true --run-ipc \
--gold-dirs=$gold_dir/0.14.1 \
--gold-dirs=$gold_dir/0.17.1 \
--gold-dirs=$gold_dir/1.0.0-bigendian \
--gold-dirs=$gold_dir/1.0.0-littleendian \
--gold-dirs=$gold_dir/2.0.0-compression \
--gold-dirs=$gold_dir/4.0.0-shareddict
```
---
docker-compose.yml | 3 ++
src/nanoarrow/ipc/decoder.c | 16 ++++----
src/nanoarrow/ipc/decoder_test.cc | 5 ++-
src/nanoarrow/ipc/reader.c | 83 ++++++++++++++++++++++++++++++++++++---
src/nanoarrow/nanoarrow_ipc.h | 5 +++
5 files changed, 98 insertions(+), 14 deletions(-)
diff --git a/docker-compose.yml b/docker-compose.yml
index bd651cb7..be7c5ada 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -55,6 +55,9 @@ services:
- ${NANOARROW_DOCKER_SOURCE_DIR}:/arrow-integration/nanoarrow
environment:
ARCHERY_INTEGRATION_TARGET_IMPLEMENTATIONS: "nanoarrow"
+ # Rust writes invalid flatbuffers:
+ # https://github.com/apache/arrow-rs/issues/5052
+ ARCHERY_INTEGRATION_WITH_RUST: "0"
command:
["echo '::group::Build nanoarrow' &&
conda run --no-capture-output /arrow-integration/ci/scripts/nanoarrow_build.sh /arrow-integration /build &&
diff --git a/src/nanoarrow/ipc/decoder.c b/src/nanoarrow/ipc/decoder.c
index 96222724..d2bad28a 100644
--- a/src/nanoarrow/ipc/decoder.c
+++ b/src/nanoarrow/ipc/decoder.c
@@ -1024,12 +1024,11 @@ static inline int ArrowIpcDecoderReadHeaderPrefix(struct ArrowIpcDecoder* decode
ArrowErrorCode ArrowIpcDecoderPeekHeader(struct ArrowIpcDecoder* decoder,
struct ArrowBufferView data,
+ int32_t* prefix_size_bytes,
struct ArrowError* error) {
ArrowIpcDecoderResetHeaderInfo(decoder);
- int32_t prefix_size_bytes;
NANOARROW_RETURN_NOT_OK(
- ArrowIpcDecoderReadHeaderPrefix(decoder, &data, &prefix_size_bytes, error));
- NANOARROW_UNUSED(prefix_size_bytes);
+ ArrowIpcDecoderReadHeaderPrefix(decoder, &data, prefix_size_bytes, error));
return NANOARROW_OK;
}
@@ -1187,15 +1186,15 @@ ArrowErrorCode ArrowIpcDecoderDecodeHeader(struct ArrowIpcDecoder* decoder,
decoder->body_size_bytes = ns(Message_bodyLength(message));
switch (decoder->metadata_version) {
- case ns(MetadataVersion_V5):
case ns(MetadataVersion_V4):
+ case ns(MetadataVersion_V5):
break;
- case ns(MetadataVersion_V1):
- case ns(MetadataVersion_V2):
- case ns(MetadataVersion_V3):
ArrowErrorSet(error, "Expected metadata version V4 or V5 but found %s",
ns(MetadataVersion_name(ns(Message_version(message)))));
return EINVAL;
+ case ns(MetadataVersion_V1):
+ case ns(MetadataVersion_V2):
+ case ns(MetadataVersion_V3):
default:
ArrowErrorSet(error, "Unexpected value for Message metadata version (%d)",
decoder->metadata_version);
@@ -1286,6 +1285,9 @@ ArrowErrorCode ArrowIpcDecoderDecodeFooter(struct ArrowIpcDecoder* decoder,
data.data.as_uint8 + data.size_bytes - footer_and_size_and_magic_size;
ns(Footer_table_t) footer = ns(Footer_as_root(footer_data));
+ NANOARROW_RETURN_NOT_OK(
+ ArrowIpcDecoderDecodeSchemaHeader(decoder, ns(Footer_schema(footer)), error));
+
NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderDecodeSchemaImpl(
ns(Footer_schema(footer)), &private_data->footer.schema, error));
diff --git a/src/nanoarrow/ipc/decoder_test.cc b/src/nanoarrow/ipc/decoder_test.cc
index 88cd7d99..078fcf70 100644
--- a/src/nanoarrow/ipc/decoder_test.cc
+++ b/src/nanoarrow/ipc/decoder_test.cc
@@ -208,7 +208,10 @@ TEST(NanoarrowIpcTest, NanoarrowIpcPeekSimpleSchema) {
data.size_bytes = sizeof(kSimpleSchema);
ArrowIpcDecoderInit(&decoder);
- EXPECT_EQ(ArrowIpcDecoderPeekHeader(&decoder, data, &error), NANOARROW_OK);
+ int32_t prefix_size_bytes = 0;
+ EXPECT_EQ(ArrowIpcDecoderPeekHeader(&decoder, data, &prefix_size_bytes, &error),
+ NANOARROW_OK);
+ EXPECT_EQ(prefix_size_bytes, 8);
EXPECT_EQ(decoder.header_size_bytes, sizeof(kSimpleSchema));
EXPECT_EQ(decoder.body_size_bytes, 0);
diff --git a/src/nanoarrow/ipc/reader.c b/src/nanoarrow/ipc/reader.c
index 3b22acf8..53937b80 100644
--- a/src/nanoarrow/ipc/reader.c
+++ b/src/nanoarrow/ipc/reader.c
@@ -28,6 +28,10 @@
#define ENODATA 120
#endif
+// Sentinel value to indicate that we haven't read a message yet
+// and don't know the number of header prefix bytes to expect.
+static const int32_t kExpectedHeaderPrefixSizeNotSet = -1;
+
void ArrowIpcInputStreamMove(struct ArrowIpcInputStream* src,
struct ArrowIpcInputStream* dst) {
memcpy(dst, src, sizeof(struct ArrowIpcInputStream));
@@ -186,6 +190,7 @@ struct ArrowIpcArrayStreamReaderPrivate {
int64_t field_index;
struct ArrowBuffer header;
struct ArrowBuffer body;
+ int32_t expected_header_prefix_size;
struct ArrowError error;
};
@@ -230,6 +235,21 @@ static int ArrowIpcArrayStreamReaderNextHeader(
// propagated higher (e.g., if the stream is empty and there's no schema message)
ArrowErrorSet(&private_data->error, "No data available on stream");
return ENODATA;
+ } else if (bytes_read == 4 && private_data->expected_header_prefix_size == 4) {
+ // Special case very, very old IPC streams that used 0x00000000 as the
+ // end-of-stream indicator. We may want to remove this case at some point:
+ // https://github.com/apache/arrow-nanoarrow/issues/648
+ uint32_t last_four_bytes = 0;
+ memcpy(&last_four_bytes, private_data->header.data, sizeof(uint32_t));
+ if (last_four_bytes == 0) {
+ ArrowErrorSet(&private_data->error, "No data available on stream");
+ return ENODATA;
+ } else {
+ ArrowErrorSet(&private_data->error,
+ "Expected 0x00000000 if exactly four bytes are available at the end "
+ "of a stream");
+ return EINVAL;
+ }
} else if (bytes_read != 8) {
ArrowErrorSet(&private_data->error,
"Expected at least 8 bytes in remainder of stream");
@@ -241,17 +261,58 @@ static int ArrowIpcArrayStreamReaderNextHeader(
input_view.size_bytes = private_data->header.size_bytes;
// Use PeekHeader to fill in decoder.header_size_bytes
- NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderPeekHeader(&private_data->decoder, input_view,
- &private_data->error));
+ int32_t prefix_size_bytes = 0;
+ NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderPeekHeader(
+ &private_data->decoder, input_view, &prefix_size_bytes, &private_data->error));
+
+ // Check for a consistent header prefix size
+ if (private_data->expected_header_prefix_size != kExpectedHeaderPrefixSizeNotSet &&
+ prefix_size_bytes != private_data->expected_header_prefix_size) {
+ ArrowErrorSet(&private_data->error,
+ "Expected prefix %d prefix header bytes but found %d",
+ (int)private_data->expected_header_prefix_size, (int)prefix_size_bytes);
+ return EINVAL;
+ } else {
+ private_data->expected_header_prefix_size = prefix_size_bytes;
+ }
+
+ // Legacy streams are missing the 0xFFFFFFFF at the start of the message. The
+ // decoder can handle this; however, verification will fail because flatbuffers
+ // must be 8-byte aligned. To handle this case, we prepend the continuation
+ // token to the start of the stream and ensure that we read four fewer bytes
+ // the next time we issue a read. We may be able to remove this case in the future:
+ // https://github.com/apache/arrow-nanoarrow/issues/648
+ int64_t extra_bytes_already_read = 0;
+ if (prefix_size_bytes == 4) {
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(ArrowBufferReserve(&private_data->header, 4),
+ &private_data->error);
+ memmove(private_data->header.data + 4, private_data->header.data,
+ private_data->header.size_bytes);
+ uint32_t continuation = 0xFFFFFFFFU;
+ memcpy(private_data->header.data, &continuation, sizeof(uint32_t));
+ private_data->header.size_bytes += 4;
+ extra_bytes_already_read = 4;
+
+ input_view.data.data = private_data->header.data;
+ input_view.size_bytes = private_data->header.size_bytes;
+
+ int32_t new_prefix_size_bytes;
+ NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderPeekHeader(&private_data->decoder, input_view,
+ &new_prefix_size_bytes,
+ &private_data->error));
+ NANOARROW_DCHECK(new_prefix_size_bytes == 8);
+ }
// Read the header bytes
int64_t expected_header_bytes = private_data->decoder.header_size_bytes - 8;
NANOARROW_RETURN_NOT_OK_WITH_ERROR(
- ArrowBufferReserve(&private_data->header, expected_header_bytes),
+ ArrowBufferReserve(&private_data->header,
+ expected_header_bytes - extra_bytes_already_read),
&private_data->error);
- NANOARROW_RETURN_NOT_OK(
- private_data->input.read(&private_data->input, private_data->header.data + 8,
- expected_header_bytes, &bytes_read, &private_data->error));
+ NANOARROW_RETURN_NOT_OK(private_data->input.read(
+ &private_data->input, private_data->header.data + private_data->header.size_bytes,
+ expected_header_bytes - extra_bytes_already_read, &bytes_read,
+ &private_data->error));
private_data->header.size_bytes += bytes_read;
// Verify + decode the header
@@ -260,6 +321,15 @@ static int ArrowIpcArrayStreamReaderNextHeader(
NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderVerifyHeader(&private_data->decoder, input_view,
&private_data->error));
+ // If we have a 4-byte header prefix, make sure the metadata version is V4
+ // (Note that some V4 IPC files have an 8 byte header prefix).
+ if (prefix_size_bytes == 4 &&
+ private_data->decoder.metadata_version != NANOARROW_IPC_METADATA_VERSION_V4) {
+ ArrowErrorSet(&private_data->error,
+ "Header prefix size of four bytes is only allowed for V4 metadata");
+ return EINVAL;
+ }
+
// Don't decode the message if it's of the wrong type (because the error message
// is better communicated by the caller)
if (private_data->decoder.message_type != message_type) {
@@ -445,6 +515,7 @@ ArrowErrorCode ArrowIpcArrayStreamReaderInit(
ArrowBufferInit(&private_data->body);
private_data->out_schema.release = NULL;
ArrowIpcInputStreamMove(input_stream, &private_data->input);
+ private_data->expected_header_prefix_size = kExpectedHeaderPrefixSizeNotSet;
if (options != NULL) {
private_data->field_index = options->field_index;
diff --git a/src/nanoarrow/nanoarrow_ipc.h b/src/nanoarrow/nanoarrow_ipc.h
index ebeb3222..fcf5ef88 100644
--- a/src/nanoarrow/nanoarrow_ipc.h
+++ b/src/nanoarrow/nanoarrow_ipc.h
@@ -251,8 +251,13 @@ void ArrowIpcDecoderReset(struct ArrowIpcDecoder* decoder);
/// these bytes and returns ESPIPE if there are not enough remaining bytes in data to read
/// the entire header message, EINVAL if the first 8 bytes are not valid, ENODATA if the
/// Arrow end-of-stream indicator has been reached, or NANOARROW_OK otherwise.
+///
+/// Pre-1.0 messages were not prefixed with 0xFFFFFFFF. For these messages, a value
+/// of 4 will be placed into prefix_size_bytes; otherwise a value of 8 will be placed
+/// into prefix_size_bytes.
ArrowErrorCode ArrowIpcDecoderPeekHeader(struct ArrowIpcDecoder* decoder,
struct ArrowBufferView data,
+ int32_t* prefix_size_bytes,
struct ArrowError* error);
/// \brief Verify a message header