#include <gtest/gtest.h>
#include <iostream>
#include <memory>
#include <nlohmann/json.hpp>
#include "table/data/vectorbatch/VectorBatch.h"
#include <taskmanager/OmniRuntimeEnvironment.h>
#include "OmniOperatorJIT/core/test/util/test_util.h"
#include "test/core/operators/OutputTest.h"
#include "table/runtime/operators/window/processor/AbstractWindowAggProcessor.h"
#include "streaming/api/operators/StreamOperatorFactory.h"
#include <vector>
using json = nlohmann::json;
using namespace omnistream;
std::string Q8hopdescription = R"DELIM({
"partition": {
"partitionName": "hash",
"channelNumber": 1,
"hashFields": [
{
"fieldIndex": 0,
"fieldName": "f0",
"fieldTypeName": "BIGINT"
}
]
},
"operators": [
{
"description": {
"timeAttributeType": "TIMESTAMP_WITHOUT_TIME_ZONE(3)",
"inputTypes": [
"BIGINT",
"VARCHAR(2147483647)",
"BIGINT"
],
"window": "TUMBLE(size=[10 s])",
"timeAttributeIndex": 2,
"aggInfoList": {
"localAggregateCalls": [],
"globalAggValueTypes": [],
"indexOfCountStar": -1,
"globalAggregateCalls": [],
"globalAccTypes": []
},
"generateUpdateBefore": false,
"outputTypes": [
"BIGINT",
"VARCHAR(2147483647)",
"BIGINT",
"BIGINT"
],
"grouping": [
0,
1
],
"originDescription": null
},
"id": "org.apache.flink.table.runtime.operators.window.slicing.SlicingWindowOperator"
}
]
})DELIM";
omnistream::VectorBatch* newVectorBatchOneKeyOneValueQ8Test1()
{
auto* vbatch = new omnistream::VectorBatch(5);
std::vector<int64_t> id = {1, 1, 1, 15, 1};
std::vector<std::string> name = {"A", "A", "B", "C", "D"};
std::vector<int64_t> time = {1000, 1000, 1000, 1000, 10000};
vbatch->Append(omniruntime::TestUtil::CreateVector<int64_t>(5, id.data()));
vbatch->Append(omniruntime::TestUtil::CreateVarcharVector(name.data(), 5));
vbatch->Append(omniruntime::TestUtil::CreateVector<int64_t>(5, time.data()));
vbatch->setRowKind(0, RowKind::INSERT);
vbatch->setRowKind(1, RowKind::INSERT);
vbatch->setRowKind(2, RowKind::INSERT);
vbatch->setRowKind(3, RowKind::INSERT);
vbatch->setRowKind(4, RowKind::INSERT);
return vbatch;
}
TEST(NEXTMARKTESTQ8, DISABLED_EMPTY)
{
omnistream::VectorBatch* vbatch = newVectorBatchOneKeyOneValueQ8Test1();
std::string uniqueName = "org.apache.flink.table.runtime.operators.window.slicing.SlicingWindowOperator";
json parsedJson = json::parse(Q8hopdescription);
omnistream::OperatorConfig opConfig(
uniqueName,
"GlobalWindowAgg_By_Simple",
parsedJson["operators"][0]["inputTypes"],
parsedJson["operators"][0]["outputTypes"],
parsedJson["operators"][0]["description"]);
auto* output = new BatchOutputTest();
auto* slicingWindowOperator = dynamic_cast<SlicingWindowOperator<std::shared_ptr<RowData>, int64_t>*>(
StreamOperatorFactory::createOperatorAndCollector(opConfig, output));
auto env2 = new omnistream::RuntimeEnvironmentV2();
auto taskInfo = new TaskInformationPOD();
taskInfo->setStateBackend("HashMapStateBackend");
{
auto configPOD = taskInfo->getStreamConfigPOD();
auto operatorDesc = configPOD.getOperatorDescription();
operatorDesc.setOperatorId("deadbeefdeadbeefdeadbeefdeadbeef");
configPOD.setOperatorDescription(operatorDesc);
taskInfo->setStreamConfigPOD(configPOD);
}
env2->SetTaskStateManager(std::make_shared<omnistream::TaskStateManager>());
env2->setTaskConfiguration(*taskInfo);
StreamTaskStateInitializerImpl* initializer = new StreamTaskStateInitializerImpl(env2);
slicingWindowOperator->setup();
slicingWindowOperator->initializeState(initializer, new LongSerializer());
std::cout << "================step 1 in Q8===================" << std::endl;
slicingWindowOperator->open();
std::cout << "================step 2 in Q8===================" << std::endl;
slicingWindowOperator->processBatch(vbatch);
slicingWindowOperator->ProcessWatermark(new Watermark(-1));
slicingWindowOperator->processBatch(vbatch);
std::cout << "================step 5 in nextmark Q8===================" << std::endl;
slicingWindowOperator->ProcessWatermark(new Watermark(-1));
std::cout << "================step 6 in nextmark Q8===================" << std::endl;
BinaryRowData* binaryRowData = BinaryRowData::createBinaryRowDataWithMem(2);
binaryRowData->setLong(0, 1);
std::string_view str = "A";
binaryRowData->setStringView(1, str);
auto* timer = new TimerHeapInternalTimer<std::shared_ptr<RowData>, int64_t>(
-1,
std::shared_ptr<RowData>(binaryRowData),
1000);
slicingWindowOperator->onTimer(timer);
std::cout << "=========== print result ==========" << std::endl;
auto* resultBatch = reinterpret_cast<omnistream::VectorBatch*>(output->getVectorBatch());
std::vector types = {"BIGINT", "VARCHAR", "BIGINT", "BIGINT"};
int rowCount = resultBatch->GetRowCount();
int colCount = resultBatch->GetVectorCount();
for (int i = 0; i < rowCount; i++) {
for (int j = 0; j < colCount; j++) {
if (types[j] == "VARCHAR") {
auto result = reinterpret_cast<
omniruntime::vec::Vector<omniruntime::vec::LargeStringContainer<std::string_view>>*>(
resultBatch->Get(j))
->GetValue(i);
std::string resStr(result);
std::cout << resStr;
} else if (types[j] == "INTEGER") {
int result = resultBatch->GetValueAt<int32_t>(j, i);
std::cout << result;
} else if (types[j] == "BIGINT") {
long result = resultBatch->GetValueAt<int64_t>(j, i);
std::cout << result;
} else {
std::string result = "NNNNO";
std::cout << result;
}
std::cout << " ";
}
std::cout << to_string(resultBatch->getRowKind(i)) << std::endl;
}
}