src/paimon/common/data/blob_test.cpp

/*
 * Copyright 2024-present Alibaba Inc.
 *
 * Licensed 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 "paimon/data/blob.h"

#include "gtest/gtest.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"

namespace paimon::test {

class BlobTest : public ::testing::Test {
 public:
    void SetUp() override {
        pool_ = GetDefaultPool();
        dir_ = UniqueTestDirectory::Create();
        ASSERT_TRUE(dir_);
        uri_ = dir_->Str() + "/file.blob";
        file_system_ = std::make_shared<LocalFileSystem>();
        ASSERT_OK(file_system_->WriteFile(uri_, "abcdefghijklmn", true));
    }
    void TearDown() override {}

 private:
    std::shared_ptr<MemoryPool> pool_;
    std::unique_ptr<UniqueTestDirectory> dir_;
    std::shared_ptr<FileSystem> file_system_;
    std::string uri_;
};

TEST_F(BlobTest, TestSimple) {
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob, Blob::FromPath(uri_));
    auto serialized = blob->ToDescriptor(pool_);

    ASSERT_OK_AND_ASSIGN(auto restored_blob,
                         Blob::FromDescriptor(serialized->data(), serialized->size()));
    ASSERT_EQ(*restored_blob->ToDescriptor(pool_), *serialized);
    ASSERT_OK_AND_ASSIGN(auto input_stream, restored_blob->NewInputStream(file_system_));
    ASSERT_EQ(uri_, restored_blob->Uri());
    std::string str(14, '\0');
    ASSERT_OK(input_stream->Read(str.data(), str.size()));
    ASSERT_EQ("abcdefghijklmn", str);
}

TEST_F(BlobTest, TestInvalidParameters) {
    // Test null file system in NewInputStream
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob, Blob::FromPath(uri_));
    ASSERT_NOK_WITH_MSG(blob->NewInputStream(nullptr), "file system is nullptr");

    // Test invalid descriptor
    std::string invalid_bytes = "invalid_descriptor_bytes";
    ASSERT_NOK(Blob::FromDescriptor(invalid_bytes.data(), invalid_bytes.size()));
}

TEST_F(BlobTest, TestRoundTripConsistency) {
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> original_blob,
                         Blob::FromPath(uri_, /*offset=*/1, /*length=*/5));

    auto first_descriptor = original_blob->ToDescriptor(pool_);
    ASSERT_OK_AND_ASSIGN(auto restored_blob,
                         Blob::FromDescriptor(first_descriptor->data(), first_descriptor->size()));

    auto second_descriptor = restored_blob->ToDescriptor(pool_);
    ASSERT_EQ(*first_descriptor, *second_descriptor);

    // Verify scheme consistency
    ASSERT_EQ(original_blob->Uri(), restored_blob->Uri());
}

TEST_F(BlobTest, TestInputStreamCreation) {
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob, Blob::FromPath(uri_));
    ASSERT_OK_AND_ASSIGN(auto input_stream, blob->NewInputStream(file_system_));
    ASSERT_TRUE(input_stream);

    // Test reading from the input stream
    std::string buffer(14, '\0');
    ASSERT_OK_AND_ASSIGN(auto bytes_read, input_stream->Read(buffer.data(), buffer.size()));
    ASSERT_EQ(14, bytes_read);
    ASSERT_EQ("abcdefghijklmn", buffer);
}

TEST_F(BlobTest, TestBoundaryConditions) {
    // Test with empty path
    ASSERT_NOK_WITH_MSG(Blob::FromPath(""), "path is an empty string");

    // Test with very large offset and length values
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob,
                         Blob::FromPath(uri_, /*offset=*/INT64_MAX, /*length=*/1));
    auto serialized = blob->ToDescriptor(pool_);

    ASSERT_OK_AND_ASSIGN(auto restored_blob,
                         Blob::FromDescriptor(serialized->data(), serialized->size()));
    ASSERT_EQ(*restored_blob->ToDescriptor(pool_), *serialized);
}

TEST_F(BlobTest, TestNewInputStreamWithOffsetAndLength) {
    // Create blob with offset=2, length=6 (should read "cdefgh")
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob,
                         Blob::FromPath(uri_, /*offset=*/2, /*length=*/6));
    ASSERT_OK_AND_ASSIGN(auto input_stream, blob->NewInputStream(file_system_));
    ASSERT_TRUE(input_stream);

    ASSERT_OK_AND_ASSIGN(uint64_t length, input_stream->Length());
    ASSERT_EQ(6, length);

    // Test reading with offset and length applied
    std::string buffer(6, '\0');
    ASSERT_OK_AND_ASSIGN(auto bytes_read, input_stream->Read(buffer.data(), buffer.size()));
    ASSERT_EQ(6, bytes_read);
    ASSERT_EQ("cdefgh", buffer);
}

TEST_F(BlobTest, TestNewInputStreamWithDynamicLength) {
    // Create blob with offset=2, dynamic length (-1 means read to end)
    ASSERT_OK_AND_ASSIGN(std::unique_ptr<Blob> blob,
                         Blob::FromPath(uri_, /*offset=*/2, /*length=*/-1));
    ASSERT_OK_AND_ASSIGN(auto input_stream, blob->NewInputStream(file_system_));
    ASSERT_TRUE(input_stream);

    ASSERT_OK_AND_ASSIGN(uint64_t length, input_stream->Length());
    ASSERT_EQ(12, length);

    // Test reading from offset to end (should read "cdefghijklmn")
    std::string buffer(12, '\0');
    ASSERT_OK_AND_ASSIGN(auto bytes_read, input_stream->Read(buffer.data(), buffer.size()));
    ASSERT_EQ(12, bytes_read);
    ASSERT_EQ("cdefghijklmn", buffer);
}

}  // namespace paimon::test