// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include <crc32c/crc32c.h>

#include "common/cast_set.h"

#include "common/logging.h"

#include "orc/Exceptions.hh"

#include "storage/utils.h"

#include "util/decompressor.h"

namespace orc {

/**

 * Decompress the bytes in to the output buffer.

 * @param inputAddress the start of the input

 * @param inputLimit one past the last byte of the input

 * @param outputAddress the start of the output buffer

 * @param outputLimit one past the last byte of the output buffer

 * @result the number of bytes decompressed

 */

uint64_t lzoDecompress(const char* inputAddress, const char* inputLimit, char* outputAddress,

                       char* outputLimit);

} // namespace orc

namespace doris {

// Lzop

const uint8_t LzopDecompressor::LZOP_MAGIC[9] = {0x89, 0x4c, 0x5a, 0x4f, 0x00,

                                                 0x0d, 0x0a, 0x1a, 0x0a};

const uint64_t LzopDecompressor::LZOP_VERSION = 0x1040;

const uint64_t LzopDecompressor::MIN_LZO_VERSION = 0x0100;

// magic(9) + ver(2) + lib_ver(2) + ver_needed(2) + method(1)

// + lvl(1) + flags(4) + mode/mtime(12) + filename_len(1)

// without the real file name, extra field and checksum

const uint32_t LzopDecompressor::MIN_HEADER_SIZE = 34;

const uint32_t LzopDecompressor::LZO_MAX_BLOCK_SIZE = (64 * 1024l * 1024l);

const uint32_t LzopDecompressor::CRC32_INIT_VALUE = 0;

const uint32_t LzopDecompressor::ADLER32_INIT_VALUE = 1;

const uint64_t LzopDecompressor::F_H_CRC32 = 0x00001000L;

const uint64_t LzopDecompressor::F_MASK = 0x00003FFFL;

const uint64_t LzopDecompressor::F_OS_MASK = 0xff000000L;

const uint64_t LzopDecompressor::F_CS_MASK = 0x00f00000L;

const uint64_t LzopDecompressor::F_RESERVED = ((F_MASK | F_OS_MASK | F_CS_MASK) ^ 0xffffffffL);

const uint64_t LzopDecompressor::F_MULTIPART = 0x00000400L;

const uint64_t LzopDecompressor::F_H_FILTER = 0x00000800L;

const uint64_t LzopDecompressor::F_H_EXTRA_FIELD = 0x00000040L;

const uint64_t LzopDecompressor::F_CRC32_C = 0x00000200L;

const uint64_t LzopDecompressor::F_ADLER32_C = 0x00000002L;

const uint64_t LzopDecompressor::F_CRC32_D = 0x00000100L;

const uint64_t LzopDecompressor::F_ADLER32_D = 0x00000001L;

Status LzopDecompressor::init() {

    return Status::OK();

}

Status LzopDecompressor::decompress(uint8_t* input, uint32_t input_len, size_t* input_bytes_read,

                                    uint8_t* output, uint32_t output_max_len,

                                    size_t* decompressed_len, bool* stream_end,

                                    size_t* more_input_bytes, size_t* more_output_bytes) {

    if (!_is_header_loaded) {

        // this is the first time to call lzo decompress, parse the header info first

        RETURN_IF_ERROR(parse_header_info(input, input_len, input_bytes_read, more_input_bytes));

        if (*more_input_bytes > 0) {

            return Status::OK();

        }

    }

    // read compressed block

    // compressed-block ::=

    //   <uncompressed-size>

    //   <compressed-size>

    //   <uncompressed-checksums>

    //   <compressed-checksums>

    //   <compressed-data>

    size_t left_input_len = input_len - *input_bytes_read;

    if (left_input_len < sizeof(uint32_t)) {

        // block is at least have uncompressed_size

        *more_input_bytes = sizeof(uint32_t) - left_input_len;

        return Status::OK();

    }

    uint8_t* block_start = input + *input_bytes_read;

    uint8_t* ptr = block_start;

    // 1. uncompressed size

    uint32_t uncompressed_size;

    ptr = get_uint32(ptr, &uncompressed_size);

    left_input_len -= sizeof(uint32_t);

    if (uncompressed_size == 0) {

        *input_bytes_read += sizeof(uint32_t);

        *stream_end = true;

        return Status::OK();

    }

    // 2. compressed size

    if (left_input_len < sizeof(uint32_t)) {

        *more_input_bytes = sizeof(uint32_t) - left_input_len;

        return Status::OK();

    }

    uint32_t compressed_size;

    ptr = get_uint32(ptr, &compressed_size);

    left_input_len -= sizeof(uint32_t);

    if (compressed_size > LZO_MAX_BLOCK_SIZE) {

        std::stringstream ss;

        ss << "lzo block size: " << compressed_size

           << " is greater than LZO_MAX_BLOCK_SIZE: " << LZO_MAX_BLOCK_SIZE;

        return Status::InternalError(ss.str());

    }

    // 3. out checksum

    uint32_t out_checksum = 0;

    if (_header_info.output_checksum_type != CHECK_NONE) {

        if (left_input_len < sizeof(uint32_t)) {

            *more_input_bytes = sizeof(uint32_t) - left_input_len;

            return Status::OK();

        }

        ptr = get_uint32(ptr, &out_checksum);

        left_input_len -= sizeof(uint32_t);

    }

    // 4. in checksum

    uint32_t in_checksum = 0;

    if (compressed_size < uncompressed_size && _header_info.input_checksum_type != CHECK_NONE) {

        if (left_input_len < sizeof(uint32_t)) {

            *more_input_bytes = sizeof(uint32_t) - left_input_len;

            return Status::OK();

        }

        ptr = get_uint32(ptr, &in_checksum);

        left_input_len -= sizeof(uint32_t);

    } else {

        // If the compressed data size is equal to the uncompressed data size, then

        // the uncompressed data is stored and there is no compressed checksum.

        in_checksum = out_checksum;

    }

    // 5. checksum compressed data

    if (left_input_len < compressed_size) {

        *more_input_bytes = compressed_size - left_input_len;

        return Status::OK();

    }

    RETURN_IF_ERROR(checksum(_header_info.input_checksum_type, "compressed", in_checksum, ptr,

                             compressed_size));

    // 6. decompress

    if (output_max_len < uncompressed_size) {

        *more_output_bytes = uncompressed_size - output_max_len;

        return Status::OK();

    }

    if (compressed_size == uncompressed_size) {

        // the data is uncompressed, just copy to the output buf

        memmove(output, ptr, compressed_size);

        ptr += compressed_size;

    } else {

        try {

            *decompressed_len =

                    orc::lzoDecompress((const char*)ptr, (const char*)(ptr + compressed_size),

                                       (char*)output, (char*)(output + uncompressed_size));

        } catch (const orc::ParseError& err) {

            std::stringstream ss;

            ss << "Lzo decompression failed: " << err.what();

            return Status::InternalError(ss.str());

        }

        RETURN_IF_ERROR(checksum(_header_info.output_checksum_type, "decompressed", out_checksum,

                                 output, *decompressed_len));

        ptr += compressed_size;

    }

    // 7. done

    *stream_end = true;

    *decompressed_len = uncompressed_size;

    *input_bytes_read += ptr - block_start;

    VLOG_DEBUG << "finished decompress lzo block."

               << " compressed_size: " << compressed_size

               << " decompressed_len: " << *decompressed_len

               << " input_bytes_read: " << *input_bytes_read;

    return Status::OK();

}

// file-header ::=  -- most of this information is not used.

//   <magic>

//   <version>

//   <lib-version>

//   [<version-needed>] -- present for all modern files.

//   <method>

//   <level>

//   <flags>

//   <mode>

//   <mtime>

//   <file-name>

//   <header-checksum>

//   <extra-field> -- presence indicated in flags, not currently used.

Status LzopDecompressor::parse_header_info(uint8_t* input, size_t input_len,

                                           size_t* input_bytes_read, size_t* more_input_bytes) {

    if (input_len < MIN_HEADER_SIZE) {

        VLOG_NOTICE << "highly recommanded that Lzo header size is larger than " << MIN_HEADER_SIZE

                    << ", or parsing header info may failed."

                    << " only given: " << input_len;

        *more_input_bytes = MIN_HEADER_SIZE - input_len;

        return Status::OK();

    }

    uint8_t* ptr = input;

    // 1. magic

    if (memcmp(ptr, LZOP_MAGIC, sizeof(LZOP_MAGIC))) {

        std::stringstream ss;

        ss << "invalid lzo magic number";

        return Status::InternalError(ss.str());

    }

    ptr += sizeof(LZOP_MAGIC);

    uint8_t* header = ptr;

    // 2. version

    ptr = get_uint16(ptr, &_header_info.version);

    if (_header_info.version > LZOP_VERSION) {

        std::stringstream ss;

        ss << "compressed with later version of lzop: " << &_header_info.version

           << " must be less than: " << LZOP_VERSION;

        return Status::InternalError(ss.str());

    }

    // 3. lib version

    ptr = get_uint16(ptr, &_header_info.lib_version);

    if (_header_info.lib_version < MIN_LZO_VERSION) {

        std::stringstream ss;

        ss << "compressed with incompatible lzo version: " << &_header_info.lib_version

           << "must be at least: " << MIN_LZO_VERSION;

        return Status::InternalError(ss.str());

    }

    // 4. version needed

    ptr = get_uint16(ptr, &_header_info.version_needed);

    if (_header_info.version_needed > LZOP_VERSION) {

        std::stringstream ss;

        ss << "compressed with imp incompatible lzo version: " << &_header_info.version

           << " must be at no more than: " << LZOP_VERSION;

        return Status::InternalError(ss.str());

    }

    // 5. method

    ptr = get_uint8(ptr, &_header_info.method);

    if (_header_info.method < 1 || _header_info.method > 3) {

        std::stringstream ss;

        ss << "invalid compression method: " << _header_info.method;

        return Status::InternalError(ss.str());

    }

    // 6. unsupported level: 7, 8, 9

    uint8_t level;

    ptr = get_uint8(ptr, &level);

    if (level > 6) {

        std::stringstream ss;

        ss << "unsupported lzo level: " << level;

        return Status::InternalError(ss.str());

    }

    // 7. flags

    uint32_t flags;

    ptr = get_uint32(ptr, &flags);

    if (flags & (F_RESERVED | F_MULTIPART | F_H_FILTER)) {

        std::stringstream ss;

        ss << "unsupported lzo flags: " << flags;

        return Status::InternalError(ss.str());

    }

    _header_info.header_checksum_type = header_type(flags);

    _header_info.input_checksum_type = input_type(flags);

    _header_info.output_checksum_type = output_type(flags);

    // 8. skip mode and mtime

    ptr += 3 * sizeof(int32_t);

    // 9. filename

    uint8_t filename_len;

    ptr = get_uint8(ptr, &filename_len);

    // here we already consume (MIN_HEADER_SIZE)

    // from now we have to check left input is enough for each step

    size_t left = input_len - (ptr - input);

    if (left < filename_len) {

        *more_input_bytes = filename_len - left;

        return Status::OK();

    }

    _header_info.filename = std::string((char*)ptr, (size_t)filename_len);

    ptr += filename_len;

    left -= filename_len;

    // 10. checksum

    if (left < sizeof(uint32_t)) {

        *more_input_bytes = sizeof(uint32_t) - left;

        return Status::OK();

    }

    uint32_t expected_checksum;

    uint8_t* cur = ptr;

    ptr = get_uint32(ptr, &expected_checksum);

    uint32_t computed_checksum;

    if (_header_info.header_checksum_type == CHECK_CRC32) {

        computed_checksum = CRC32_INIT_VALUE;

        computed_checksum = crc32c::Extend(computed_checksum, (const uint8_t*)header, cur - header);

    } else {

        computed_checksum = ADLER32_INIT_VALUE;

        computed_checksum = olap_adler32(computed_checksum, (const char*)header, cur - header);

    }

    if (computed_checksum != expected_checksum) {

        std::stringstream ss;

        ss << "invalid header checksum: " << computed_checksum

           << " expected: " << expected_checksum;

        return Status::InternalError(ss.str());

    }

    left -= sizeof(uint32_t);

    // 11. skip extra

    if (flags & F_H_EXTRA_FIELD) {

        if (left < sizeof(uint32_t)) {

            *more_input_bytes = sizeof(uint32_t) - left;

            return Status::OK();

        }

        uint32_t extra_len;

        ptr = get_uint32(ptr, &extra_len);

        left -= sizeof(uint32_t);

        // add the checksum and the len to the total ptr size.

        if (left < sizeof(int32_t) + extra_len) {

            *more_input_bytes = sizeof(int32_t) + extra_len - left;

            return Status::OK();

        }

        left -= sizeof(int32_t) + extra_len;

        ptr += sizeof(int32_t) + extra_len;

    }

    _header_info.header_size = cast_set<int32_t>(ptr - input);

    *input_bytes_read = _header_info.header_size;

    _is_header_loaded = true;

    VLOG_DEBUG << debug_info();

    return Status::OK();

}

Status LzopDecompressor::checksum(LzoChecksum type, const std::string& source, uint32_t expected,

                                  uint8_t* ptr, size_t len) {

    uint32_t computed_checksum;

    switch (type) {

    case CHECK_NONE:

        return Status::OK();

    case CHECK_CRC32:

        computed_checksum = crc32c::Extend(CRC32_INIT_VALUE, (const uint8_t*)ptr, len);

        break;

    case CHECK_ADLER:

        computed_checksum = olap_adler32(ADLER32_INIT_VALUE, (const char*)ptr, len);

        break;

    default:

        std::stringstream ss;

        ss << "Invalid checksum type: " << type;

        return Status::InternalError(ss.str());

    }

    if (computed_checksum != expected) {

        std::stringstream ss;

        ss << "checksum of " << source << " block failed."

           << " computed checksum: " << computed_checksum << " expected: " << expected;

        return Status::InternalError(ss.str());

    }

    return Status::OK();

}

std::string LzopDecompressor::debug_info() {

    std::stringstream ss;

    ss << "LzopDecompressor."

       << " version: " << _header_info.version << " lib version: " << _header_info.lib_version

       << " version needed: " << _header_info.version_needed

       << " method: " << (uint16_t)_header_info.method << " filename: " << _header_info.filename

       << " header size: " << _header_info.header_size

       << " header checksum type: " << _header_info.header_checksum_type

       << " input checksum type: " << _header_info.input_checksum_type

       << " output checksum type: " << _header_info.output_checksum_type;

    return ss.str();

}

} // namespace doris