paleolimbot commented on code in PR #935:
URL: https://github.com/apache/arrow-nanoarrow/pull/935#discussion_r3960838362
##########
src/nanoarrow/ipc/encoder.c:
##########
@@ -49,6 +49,16 @@ struct ArrowIpcEncoderPrivate {
// Metadata to attach to the next encoded Message (in nanoarrow's packed
// representation), or an empty buffer if the next Message has no metadata.
struct ArrowBuffer message_metadata;
+ // Compression applied to the body buffers of subsequently encoded
RecordBatches
+ enum ArrowIpcCompressionType codec;
+ // Compressor used when codec != NONE (release is NULL until one is needed)
+ struct ArrowIpcCompressor compressor;
+ // Whether compressor was provided by ArrowIpcEncoderSetCompressor()
+ int custom_compressor;
+ // Compression level passed to the compressor when codec != NONE
+ int compression_level;
+ // The flatbuffer equivalent of codec (only meaningful when codec != NONE)
+ ns(CompressionType_enum_t) flatbuf_codec;
Review Comment:
Can we reduce this to only `codec` and `compressor` or even just
`compressor` by inlining the codec and/or level into the `ArrowIpcCompressor`
struct?
##########
src/nanoarrow/ipc/codecs.c:
##########
@@ -113,6 +183,99 @@ ArrowIpcDecompressFunction
ArrowIpcGetLZ4DecompressionFunction(void) {
#endif
}
+ArrowIpcCompressFunction ArrowIpcGetLZ4CompressionFunction(void) {
+#if defined(NANOARROW_IPC_WITH_LZ4)
+ return &ArrowIpcCompressLZ4;
+#else
+ return NULL;
+#endif
+}
+
+ArrowErrorCode ArrowIpcGetCompressionLevelRange(
+ enum ArrowIpcCompressionType compression_type, int* min_level_out,
+ int* max_level_out) {
+ NANOARROW_DCHECK(min_level_out != NULL && max_level_out != NULL);
+ NANOARROW_UNUSED(min_level_out);
+ NANOARROW_UNUSED(max_level_out);
+
+ switch (compression_type) {
+ case NANOARROW_IPC_COMPRESSION_TYPE_ZSTD:
+#if defined(NANOARROW_IPC_WITH_ZSTD)
+#if ZSTD_VERSION_NUMBER >= 10400
+ *min_level_out = ZSTD_minCLevel();
+#else
+ // Negative (fast) levels can't be queried before zstd 1.4.0
+ *min_level_out = NANOARROW_IPC_COMPRESSION_LEVEL_DEFAULT;
+#endif
+ *max_level_out = ZSTD_maxCLevel();
+ return NANOARROW_OK;
+#else
+ return ENOTSUP;
+#endif
+ case NANOARROW_IPC_COMPRESSION_TYPE_LZ4_FRAME:
+#if defined(NANOARROW_IPC_WITH_LZ4)
+ // A negative level selects an acceleration of 1 - level, which lz4 caps
at 65537
+ // (LZ4_ACCELERATION_MAX, which is not part of its public headers)
+ *min_level_out = 1 - 65537;
+ *max_level_out = LZ4F_compressionLevel_max();
+ return NANOARROW_OK;
+#else
+ return ENOTSUP;
+#endif
+ default:
+ return EINVAL;
+ }
+}
+
+// The serial decompressor and compressor keep one function per codec, indexed
by
+// enum ArrowIpcCompressionType (NONE is never a codec)
+static int ArrowIpcCompressionTypeIsCodec(enum ArrowIpcCompressionType
compression_type) {
+ switch (compression_type) {
+ case NANOARROW_IPC_COMPRESSION_TYPE_ZSTD:
+ case NANOARROW_IPC_COMPRESSION_TYPE_LZ4_FRAME:
+ return 1;
+ default:
+ return 0;
+ }
+}
+
+const char* ArrowIpcCompressionTypeToString(
+ enum ArrowIpcCompressionType compression_type) {
+ switch (compression_type) {
+ case NANOARROW_IPC_COMPRESSION_TYPE_NONE:
+ return "none";
+ case NANOARROW_IPC_COMPRESSION_TYPE_LZ4_FRAME:
+ return "lz4";
+ case NANOARROW_IPC_COMPRESSION_TYPE_ZSTD:
+ return "zstd";
+ default:
+ return NULL;
Review Comment:
It's better to have this return `""` or `"<unknown>"` for the inevitable
user or contributor who uses this in a format string without checking for a
NULL. I had to fix ArrowTypeToString recently because of gcc warnings.
##########
src/nanoarrow/nanoarrow_ipc.h:
##########
@@ -412,6 +432,102 @@ ArrowIpcSerialDecompressorSetFunction(struct
ArrowIpcDecompressor* decompressor,
enum ArrowIpcCompressionType
compression_type,
ArrowIpcDecompressFunction
decompress_function);
+/// \brief Compression level that selects the codec's default level
+#define NANOARROW_IPC_COMPRESSION_LEVEL_DEFAULT 0
+
+/// \brief A user-extensible compressor
+///
+/// The ArrowIpcCompressor is the underlying object that enables buffer
compression
+/// in the ArrowIpcEncoder. An implementation of a compressor may support more
than one
+/// ArrowIpcCompressionType.
+struct ArrowIpcCompressor {
+ /// \brief Compress a buffer
+ ///
+ /// Compresses src using compression_type at compression_level and appends
the
+ /// compressed bytes to dst. Any content already in dst must be preserved
(i.e.,
+ /// implementations may only append to dst). See ArrowIpcCompressFunction
for the
+ /// interpretation of compression_level.
+ ArrowErrorCode (*compress)(struct ArrowIpcCompressor* compressor,
+ enum ArrowIpcCompressionType compression_type,
+ int compression_level, struct ArrowBufferView src,
+ struct ArrowBuffer* dst, struct ArrowError*
error);
+
+ /// \brief Release the compressor and any resources it may be holding
+ ///
+ /// Release callback implementations must set the release member to NULL.
+ /// Callers must check that the release callback is not NULL before calling
+ /// compress() or release().
+ void (*release)(struct ArrowIpcCompressor* compressor);
+
+ /// \brief Implementation-specific opaque data
+ void* private_data;
+};
+
+/// \brief A self-contained compression function
+///
+/// Compresses src at compression_level and appends the compressed bytes to
dst. Because
+/// the compressed size is not known in advance, implementations are
responsible for
+/// reserving sufficient space in dst (e.g., using the compression library's
bound
+/// function) and must only append to dst.
+///
+/// The interpretation of compression_level is codec-specific:
+/// NANOARROW_IPC_COMPRESSION_LEVEL_DEFAULT selects the codec's default level
and other
+/// values follow the underlying library's conventions (see
+/// ArrowIpcGetCompressionLevelRange()). ArrowIpcEncoderSetCompression()
rejects levels
+/// outside that range; the built-in functions clamp them if called directly.
+typedef ArrowErrorCode (*ArrowIpcCompressFunction)(struct ArrowBufferView src,
+ int compression_level,
+ struct ArrowBuffer* dst,
+ struct ArrowError* error);
+
+/// \brief Get the compression function for ZSTD
+///
+/// The result will be NULL if nanoarrow was not built with
NANOARROW_IPC_WITH_ZSTD.
+NANOARROW_DLL ArrowIpcCompressFunction
ArrowIpcGetZstdCompressionFunction(void);
+
+/// \brief Get the compression function for LZ4
+///
+/// The result will be NULL if nanoarrow was not built with
NANOARROW_IPC_WITH_LZ4.
+NANOARROW_DLL ArrowIpcCompressFunction ArrowIpcGetLZ4CompressionFunction(void);
+
+/// \brief An ArrowIpcCompressor implementation that performs compression in
serial
+NANOARROW_DLL ArrowErrorCode
+ArrowIpcSerialCompressor(struct ArrowIpcCompressor* compressor);
Review Comment:
I think this can take a `compression_type` and `level` parameter if we
inline those parameters into the ArrowIpcSerialCompressor private data.
##########
src/nanoarrow/nanoarrow_ipc.h:
##########
@@ -412,6 +432,102 @@ ArrowIpcSerialDecompressorSetFunction(struct
ArrowIpcDecompressor* decompressor,
enum ArrowIpcCompressionType
compression_type,
ArrowIpcDecompressFunction
decompress_function);
+/// \brief Compression level that selects the codec's default level
+#define NANOARROW_IPC_COMPRESSION_LEVEL_DEFAULT 0
+
+/// \brief A user-extensible compressor
+///
+/// The ArrowIpcCompressor is the underlying object that enables buffer
compression
+/// in the ArrowIpcEncoder. An implementation of a compressor may support more
than one
+/// ArrowIpcCompressionType.
+struct ArrowIpcCompressor {
+ /// \brief Compress a buffer
+ ///
+ /// Compresses src using compression_type at compression_level and appends
the
+ /// compressed bytes to dst. Any content already in dst must be preserved
(i.e.,
+ /// implementations may only append to dst). See ArrowIpcCompressFunction
for the
+ /// interpretation of compression_level.
+ ArrowErrorCode (*compress)(struct ArrowIpcCompressor* compressor,
+ enum ArrowIpcCompressionType compression_type,
+ int compression_level, struct ArrowBufferView src,
+ struct ArrowBuffer* dst, struct ArrowError*
error);
Review Comment:
I think it would be better for force any parameters (like level and
compression type) to be captured in the private data, perhaps with a member for
compression_type since the encoder will need to declare it (callback seems like
overkill for that).
If there is any way to potentially parallelize compression, it would be
great to have this struct/its use support that even though the default
implementation won't. You might need two passes (`compress_start()`,
`compress_finish()`, passing `i` to identify the buffer). I don't want to spend
too much time on the best possible parallel encoding, I just want to avoid
changing the ABI for a trivial two pass setup. I don't mind the exact contract
as long as there's at least one plausible setup where this works. Also fine if
there is no easy plausible setup (we can try later pre 1.0).
##########
src/nanoarrow/ipc/encoder.c:
##########
@@ -69,6 +79,11 @@ ArrowErrorCode ArrowIpcEncoderInit(struct ArrowIpcEncoder*
encoder) {
ArrowBufferInit(&private->nodes);
ArrowIpcDictionaryEncodingsInit(&private->dictionary_encodings);
ArrowBufferInit(&private->message_metadata);
+ private->codec = NANOARROW_IPC_COMPRESSION_TYPE_NONE;
+ private->compressor.release = NULL;
+ private->custom_compressor = 0;
+ private->compression_level = NANOARROW_IPC_COMPRESSION_LEVEL_DEFAULT;
+ private->flatbuf_codec = ns(CompressionType_LZ4_FRAME);
Review Comment:
Is there another value we can use for `flatbuf_codec` (or can we omit the
field entirely and recalculate it when needed) that better communicates
defaultness (rather than a specific value)?
##########
src/nanoarrow/ipc/encoder.c:
##########
@@ -744,6 +886,15 @@ static ArrowErrorCode ArrowIpcEncoderEncodeRecordBatch(
FLATCC_RETURN_UNLESS_0(Message_header_RecordBatch_start(builder), error);
FLATCC_RETURN_UNLESS_0(RecordBatch_length_add(builder, array_view->length),
error);
+ if (private->codec != NANOARROW_IPC_COMPRESSION_TYPE_NONE) {
+ FLATCC_RETURN_UNLESS_0(RecordBatch_compression_start(builder), error);
+ FLATCC_RETURN_UNLESS_0(BodyCompression_codec_add(builder,
private->flatbuf_codec),
+ error);
+ FLATCC_RETURN_UNLESS_0(
+ BodyCompression_method_add(builder, ns(BodyCompressionMethod_BUFFER)),
error);
+ FLATCC_RETURN_UNLESS_0(RecordBatch_compression_end(builder), error);
+ }
+
Review Comment:
In addition to these changes, I believe the use of compression should also
be declared when writing the Schema message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
