duckdb::MiniZStreamWrapper Struct Reference

Inheritance diagram for duckdb::MiniZStreamWrapper:

Collaboration diagram for duckdb::MiniZStreamWrapper:

Public Member Functions
void	Initialize (QueryContext context, CompressedFile &file, bool write) override
bool	Read (StreamData &stream_data) override
void	Write (CompressedFile &file, StreamData &stream_data, data_ptr_t buffer, int64_t nr_bytes) override
void	Close () override
void	FlushStream () const

Public Attributes
CompressedFile *	file = nullptr
unique_ptr< duckdb_miniz::mz_stream >	mz_stream_ptr
bool	writing = false
duckdb_miniz::mz_ulong	crc
idx_t	total_size

Constructor & Destructor Documentation

~MiniZStreamWrapper()

duckdb::MiniZStreamWrapper::~MiniZStreamWrapper ( )

override

                                        {
    // avoid closing if destroyed during stack unwinding
    if (Exception::UncaughtException()) {
        return;
    }
    try {
        MiniZStreamWrapper::Close();
    } catch (std::exception &ex) {
        if (file && file->child_handle) {
            // FIXME: Make any log context available here.
            ErrorData data(ex);
            try {
                const auto logger = file->child_handle->logger;
                if (logger) {
                    DUCKDB_LOG_ERROR(logger, "MiniZStreamWrapper::~MiniZStreamWrapper()\t\t" + data.Message())
                }
            } catch (...) { // NOLINT
            }
        }
    } catch (...) { // NOLINT
    }
}

Member Function Documentation

Initialize()

void duckdb::MiniZStreamWrapper::Initialize ( QueryContext  context, CompressedFile &  file, bool  write  )

overridevirtual

Implements duckdb::StreamWrapper.

                                                                                          {
    Close();
    this->file = &file;
    mz_stream_ptr = make_uniq<duckdb_miniz::mz_stream>();
    memset(mz_stream_ptr.get(), 0, sizeof(duckdb_miniz::mz_stream));
    this->writing = write;
 
    // TODO use custom alloc/free methods in miniz to throw exceptions on OOM
    uint8_t gzip_hdr[GZIP_HEADER_MINSIZE];
    if (write) {
        crc = MZ_CRC32_INIT;
        total_size = 0;
 
        MiniZStream::InitializeGZIPHeader(gzip_hdr);
        file.child_handle->Write(context, gzip_hdr, GZIP_HEADER_MINSIZE);
 
        auto ret = mz_deflateInit2(mz_stream_ptr.get(), duckdb_miniz::MZ_DEFAULT_LEVEL, MZ_DEFLATED,
                                   -MZ_DEFAULT_WINDOW_BITS, 1, 0);
        if (ret != duckdb_miniz::MZ_OK) {
            throw InternalException("Failed to initialize miniz");
        }
    } else {
        idx_t data_start = GZIP_HEADER_MINSIZE;
        auto read_count = file.child_handle->Read(context, gzip_hdr, GZIP_HEADER_MINSIZE);
        GZipFileSystem::VerifyGZIPHeader(gzip_hdr, NumericCast<idx_t>(read_count), &file);
        // Skip over the extra field if necessary
        if (gzip_hdr[3] & GZIP_FLAG_EXTRA) {
            uint8_t gzip_xlen[2];
            file.child_handle->Seek(data_start);
            file.child_handle->Read(context, gzip_xlen, 2);
            auto xlen = NumericCast<idx_t>((uint8_t)gzip_xlen[0] | (uint8_t)gzip_xlen[1] << 8);
            data_start += xlen + 2;
        }
        // Skip over the file name if necessary
        if (gzip_hdr[3] & GZIP_FLAG_NAME) {
            file.child_handle->Seek(data_start);
            data_start += GZipConsumeString(context, *file.child_handle);
        }
        file.child_handle->Seek(data_start);
        // stream is now set to beginning of payload data
        auto ret = duckdb_miniz::mz_inflateInit2(mz_stream_ptr.get(), -MZ_DEFAULT_WINDOW_BITS);
        if (ret != duckdb_miniz::MZ_OK) {
            throw InternalException("Failed to initialize miniz");
        }
    }
}

Read()

bool duckdb::MiniZStreamWrapper::Read ( StreamData &  stream_data )

overridevirtual

Implements duckdb::StreamWrapper.

                                            {
    // Handling for the concatenated files
    if (sd.refresh) {
        auto available = static_cast<uint32_t>(sd.in_buff_end - sd.in_buff_start);
        if (available <= GZIP_FOOTER_SIZE) {
            // Only footer is available so we just close and return finished
            Close();
            return true;
        }
 
        sd.refresh = false;
        auto body_ptr = sd.in_buff_start + GZIP_FOOTER_SIZE;
        uint8_t gzip_hdr[GZIP_HEADER_MINSIZE];
        memcpy(gzip_hdr, body_ptr, GZIP_HEADER_MINSIZE);
        GZipFileSystem::VerifyGZIPHeader(gzip_hdr, GZIP_HEADER_MINSIZE, nullptr);
        body_ptr += GZIP_HEADER_MINSIZE;
        if (gzip_hdr[3] & GZIP_FLAG_EXTRA) {
            auto xlen = NumericCast<idx_t>((uint8_t)*body_ptr | (uint8_t) * (body_ptr + 1) << 8);
            body_ptr += xlen + 2;
            if (GZIP_FOOTER_SIZE + GZIP_HEADER_MINSIZE + 2 + xlen >= GZIP_HEADER_MAXSIZE) {
                throw InternalException("Extra field resulting in GZIP header larger than defined maximum (%d)",
                                        GZIP_HEADER_MAXSIZE);
            }
        }
        if (gzip_hdr[3] & GZIP_FLAG_NAME) {
            char c;
            do {
                c = UnsafeNumericCast<char>(*body_ptr);
                body_ptr++;
            } while (c != '\0' && body_ptr < sd.in_buff_end);
            if (static_cast<idx_t>(body_ptr - sd.in_buff_start) >= GZIP_HEADER_MAXSIZE) {
                throw InternalException("Filename resulting in GZIP header larger than defined maximum (%d)",
                                        GZIP_HEADER_MAXSIZE);
            }
        }
        sd.in_buff_start = body_ptr;
        if (sd.in_buff_end - sd.in_buff_start < 1) {
            Close();
            return true;
        }
        duckdb_miniz::mz_inflateEnd(mz_stream_ptr.get());
        auto sta = duckdb_miniz::mz_inflateInit2(mz_stream_ptr.get(), -MZ_DEFAULT_WINDOW_BITS);
        if (sta != duckdb_miniz::MZ_OK) {
            throw InternalException("Failed to initialize miniz");
        }
    }
 
    // actually decompress
    mz_stream_ptr->next_in = sd.in_buff_start;
    D_ASSERT(sd.in_buff_end - sd.in_buff_start < NumericLimits<int32_t>::Maximum());
    mz_stream_ptr->avail_in = static_cast<uint32_t>(sd.in_buff_end - sd.in_buff_start);
    mz_stream_ptr->next_out = data_ptr_cast(sd.out_buff_end);
    mz_stream_ptr->avail_out = static_cast<uint32_t>((sd.out_buff.get() + sd.out_buf_size) - sd.out_buff_end);
    auto ret = duckdb_miniz::mz_inflate(mz_stream_ptr.get(), duckdb_miniz::MZ_NO_FLUSH);
    if (ret != duckdb_miniz::MZ_OK && ret != duckdb_miniz::MZ_STREAM_END) {
        throw IOException("Failed to decode gzip stream: %s", duckdb_miniz::mz_error(ret));
    }
    // update pointers following inflate()
    sd.in_buff_start = (data_ptr_t)mz_stream_ptr->next_in; // NOLINT
    sd.in_buff_end = sd.in_buff_start + mz_stream_ptr->avail_in;
    sd.out_buff_end = data_ptr_cast(mz_stream_ptr->next_out);
    D_ASSERT(sd.out_buff_end + mz_stream_ptr->avail_out == sd.out_buff.get() + sd.out_buf_size);
 
    // if stream ended, deallocate inflator
    if (ret == duckdb_miniz::MZ_STREAM_END) {
        // Concatenated GZIP potentially coming up - refresh input buffer
        sd.refresh = true;
    }
    return false;
}

Write()

void duckdb::MiniZStreamWrapper::Write ( CompressedFile &  file, StreamData &  stream_data, data_ptr_t  buffer, int64_t  nr_bytes  )

overridevirtual

Implements duckdb::StreamWrapper.

                                                          {
    // update the src and the total size
    crc = duckdb_miniz::mz_crc32(crc, reinterpret_cast<const unsigned char *>(uncompressed_data),
                                 UnsafeNumericCast<size_t>(uncompressed_size));
    total_size += UnsafeNumericCast<idx_t>(uncompressed_size);
 
    auto remaining = uncompressed_size;
    while (remaining > 0) {
        auto output_remaining = UnsafeNumericCast<idx_t>((sd.out_buff.get() + sd.out_buf_size) - sd.out_buff_start);
 
        mz_stream_ptr->next_in = reinterpret_cast<const unsigned char *>(uncompressed_data);
        mz_stream_ptr->avail_in = NumericCast<unsigned int>(remaining);
        mz_stream_ptr->next_out = sd.out_buff_start;
        mz_stream_ptr->avail_out = NumericCast<unsigned int>(output_remaining);
 
        auto res = mz_deflate(mz_stream_ptr.get(), duckdb_miniz::MZ_NO_FLUSH);
        if (res != duckdb_miniz::MZ_OK) {
            D_ASSERT(res != duckdb_miniz::MZ_STREAM_END);
            throw InternalException("Failed to compress GZIP block");
        }
        sd.out_buff_start += output_remaining - mz_stream_ptr->avail_out;
        if (mz_stream_ptr->avail_out == 0) {
            // no more output buffer available: flush
            file.child_handle->Write(sd.out_buff.get(),
                                     UnsafeNumericCast<idx_t>(sd.out_buff_start - sd.out_buff.get()));
            sd.out_buff_start = sd.out_buff.get();
        }
        auto written = UnsafeNumericCast<idx_t>(remaining - mz_stream_ptr->avail_in);
        uncompressed_data += written;
        remaining = mz_stream_ptr->avail_in;
    }
}

Close()

void duckdb::MiniZStreamWrapper::Close ( )

overridevirtual

Implements duckdb::StreamWrapper.

                               {
    if (!mz_stream_ptr) {
        return;
    }
    if (writing) {
        // flush anything remaining in the stream
        FlushStream();
 
        // write the footer
        unsigned char gzip_footer[MiniZStream::GZIP_FOOTER_SIZE];
        MiniZStream::InitializeGZIPFooter(gzip_footer, crc, total_size);
        file->child_handle->Write(gzip_footer, MiniZStream::GZIP_FOOTER_SIZE);
 
        duckdb_miniz::mz_deflateEnd(mz_stream_ptr.get());
    } else {
        duckdb_miniz::mz_inflateEnd(mz_stream_ptr.get());
    }
    mz_stream_ptr = nullptr;
    file = nullptr;
}

FlushStream()

void duckdb::MiniZStreamWrapper::FlushStream

(

)

const

                                           {
    auto &sd = file->stream_data;
    mz_stream_ptr->next_in = nullptr;
    mz_stream_ptr->avail_in = 0;
    while (true) {
        auto output_remaining = (sd.out_buff.get() + sd.out_buf_size) - sd.out_buff_start;
        mz_stream_ptr->next_out = sd.out_buff_start;
        mz_stream_ptr->avail_out = NumericCast<unsigned int>(output_remaining);
 
        auto res = mz_deflate(mz_stream_ptr.get(), duckdb_miniz::MZ_FINISH);
        sd.out_buff_start += (output_remaining - mz_stream_ptr->avail_out);
        if (sd.out_buff_start > sd.out_buff.get()) {
            file->child_handle->Write(sd.out_buff.get(),
                                      UnsafeNumericCast<idx_t>(sd.out_buff_start - sd.out_buff.get()));
            sd.out_buff_start = sd.out_buff.get();
        }
        if (res == duckdb_miniz::MZ_STREAM_END) {
            break;
        }
        if (res != duckdb_miniz::MZ_OK) {
            throw InternalException("Failed to compress GZIP block");
        }
    }
}

---

The documentation for this struct was generated from the following file:

• external/duckdb/duckdb.cpp