// 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 "exec/rowid_fetcher.h"

#include <gtest/gtest.h>

#include <string>
#include <vector>

#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"

namespace doris {

// source_column_key() decides which projected slots collapse onto a single scanned
// column. Two slots that name the same physical column must produce the same key, and
// any two slots that do not must produce different ones -- a false merge drops a column
// the result block still expects, which is the crash this keying was introduced to fix.
class RowIdStorageReaderTest : public testing::Test {
public:
    static std::string key_of(const SlotDescriptor& slot, uint32_t column_idx) {
        return RowIdStorageReader::source_column_key(slot, column_idx);
    }

protected:
    struct SlotSpec {
        std::string col_name = "c";
        int32_t col_unique_id = 1;
        std::vector<std::string> column_paths = {};
        TColumnAccessPaths access_paths = {};
    };

    static SlotDescriptor make_slot(const SlotSpec& spec) {
        TSlotDescriptor tdesc = TSlotDescriptorBuilder()
                                        .type(TYPE_INT)
                                        .nullable(true)
                                        .column_name(spec.col_name)
                                        .column_pos(0)
                                        .build();
        tdesc.__set_col_unique_id(spec.col_unique_id);
        tdesc.__set_column_paths(spec.column_paths);
        if (!spec.access_paths.empty()) {
            tdesc.__set_all_access_paths(spec.access_paths);
        }
        return SlotDescriptor(tdesc);
    }

    static TColumnAccessPath data_path(const std::vector<std::string>& path) {
        TColumnAccessPath access_path;
        access_path.type = TAccessPathType::DATA;
        TDataAccessPath data;
        data.__set_path(path);
        access_path.__set_data_access_path(data);
        return access_path;
    }

    static TColumnAccessPath bare_path() {
        TColumnAccessPath access_path;
        access_path.type = TAccessPathType::DATA;
        return access_path;
    }
};

TEST_F(RowIdStorageReaderTest, SameSourceColumnSharesKey) {
    // The bug case: one physical column projected twice must dedup onto one scan column.
    const SlotDescriptor first = make_slot({});
    const SlotDescriptor second = make_slot({});
    EXPECT_EQ(key_of(first, 3), key_of(second, 3));
}

TEST_F(RowIdStorageReaderTest, ColumnIndexSeparatesKeys) {
    const SlotDescriptor slot = make_slot({});
    EXPECT_NE(key_of(slot, 3), key_of(slot, 4));
}

TEST_F(RowIdStorageReaderTest, ColumnNameSeparatesKeys) {
    EXPECT_NE(key_of(make_slot({.col_name = "a"}), 0), key_of(make_slot({.col_name = "b"}), 0));
}

TEST_F(RowIdStorageReaderTest, UniqueIdSeparatesKeys) {
    EXPECT_NE(key_of(make_slot({.col_unique_id = 1}), 0),
              key_of(make_slot({.col_unique_id = 2}), 0));
}

TEST_F(RowIdStorageReaderTest, NameAndIndexBoundaryIsNotAmbiguous) {
    // Without length prefixes, "a" + idx 12 and "a1" + idx 2 both flatten to "a12".
    EXPECT_NE(key_of(make_slot({.col_name = "a"}), 12), key_of(make_slot({.col_name = "a1"}), 2));
}

TEST_F(RowIdStorageReaderTest, PathComponentBoundaryIsNotAmbiguous) {
    // The concatenation hazard the length prefix exists for: ["a", "b"] and ["a:b"] are
    // different nested columns but share the naive ':'-joined spelling.
    EXPECT_NE(key_of(make_slot({.column_paths = {"a", "b"}}), 0),
              key_of(make_slot({.column_paths = {"a:b"}}), 0));
}

TEST_F(RowIdStorageReaderTest, EmptyPathIsNotTheSameAsNoPath) {
    EXPECT_NE(key_of(make_slot({.column_paths = {}}), 0),
              key_of(make_slot({.column_paths = {""}}), 0));
}

TEST_F(RowIdStorageReaderTest, PathOrderMatters) {
    EXPECT_NE(key_of(make_slot({.column_paths = {"a", "b"}}), 0),
              key_of(make_slot({.column_paths = {"b", "a"}}), 0));
}

TEST_F(RowIdStorageReaderTest, EqualPathsShareKey) {
    EXPECT_EQ(key_of(make_slot({.column_paths = {"a", "b"}}), 0),
              key_of(make_slot({.column_paths = {"a", "b"}}), 0));
}

TEST_F(RowIdStorageReaderTest, AccessPathSeparatesKeys) {
    EXPECT_NE(key_of(make_slot({.access_paths = {data_path({"a"})}}), 0),
              key_of(make_slot({.access_paths = {data_path({"b"})}}), 0));
}

TEST_F(RowIdStorageReaderTest, AbsentAccessPathIsNotAnEmptyOne) {
    // The presence bit: an unset data_access_path must not collide with one that is set
    // but carries no components.
    EXPECT_NE(key_of(make_slot({.access_paths = {bare_path()}}), 0),
              key_of(make_slot({.access_paths = {data_path({})}}), 0));
}

TEST_F(RowIdStorageReaderTest, AccessPathCountSeparatesKeys) {
    EXPECT_NE(key_of(make_slot({.access_paths = {data_path({"a"})}}), 0),
              key_of(make_slot({.access_paths = {data_path({"a"}), data_path({"b"})}}), 0));
}

// Runs every submitted task on the submitting thread. The point of these cases is which
// status reaches the caller, not the threading, and inline execution keeps them
// deterministic.
class InlineScanScheduler : public ScannerScheduler {
public:
    Status start(int, int, int, int) override { return Status::OK(); }
    void stop() override {}
    Status submit_scan_task(SimplifiedScanTask scan_task) override {
        scan_task.scan_func();
        return Status::OK();
    }
    Status submit_scan_task(SimplifiedScanTask scan_task, const std::string&) override {
        scan_task.scan_func();
        return Status::OK();
    }
    void reset_thread_num(int, int, int) override {}
    int get_queue_size() override { return 0; }
    int get_active_threads() override { return 0; }
    std::vector<int> thread_debug_info() override { return {}; }
    Status schedule_scan_task(std::shared_ptr<ScannerContext>, std::shared_ptr<ScanTask>,
                              std::unique_lock<std::mutex>&) override {
        return Status::OK();
    }
};

// submit_external_scan_tasks() signals completion from a Defer, so a worker that leaves
// without publishing its status would still wake the waiter and the caller would report
// success over a partially filled result block.
class SubmitExternalScanTasksTest : public RowIdStorageReaderTest {
protected:
    static constexpr size_t kTaskCount = 3;

    static Status run_tasks(const std::function<Status(size_t)>& run_task) {
        InlineScanScheduler scheduler;
        std::counting_semaphore<> semaphore {kTaskCount};
        return RowIdStorageReader::submit_external_scan_tasks(
                &scheduler, semaphore, kTaskCount,
                [](size_t idx) { return fmt::format("task-{}", idx); }, run_task);
    }
};

TEST_F(SubmitExternalScanTasksTest, AllTasksSucceedingReturnsOk) {
    size_t ran = 0;
    EXPECT_TRUE(run_tasks([&](size_t) -> Status {
                    ++ran;
                    return Status::OK();
                }).ok());
    EXPECT_EQ(ran, kTaskCount);
}

TEST_F(SubmitExternalScanTasksTest, ReturnedErrorReachesTheCaller) {
    const Status result = run_tasks([](size_t idx) -> Status {
        return idx == kTaskCount - 1 ? Status::InternalError("scanner returned an error")
                                     : Status::OK();
    });
    EXPECT_FALSE(result.ok());
    EXPECT_NE(result.to_string().find("scanner returned an error"), std::string::npos);
}

TEST_F(SubmitExternalScanTasksTest, ThrownExceptionReachesTheCaller) {
    // The last task is the interesting one: it is the completion that releases the
    // waiter, so a status lost here is a status the caller never sees.
    const Status result = run_tasks([](size_t idx) -> Status {
        if (idx == kTaskCount - 1) {
            throw Exception(ErrorCode::INTERNAL_ERROR, "scanner threw");
        }
        return Status::OK();
    });
    EXPECT_FALSE(result.ok());
    EXPECT_NE(result.to_string().find("scanner threw"), std::string::npos);
}

} // namespace doris