yql/essentials/providers/pg/provider/yql_pg_datasink.cpp


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

#include "yql_pg_provider_impl.h"

#include <yql/essentials/providers/common/provider/yql_data_provider_impl.h>
#include <yql/essentials/providers/common/provider/yql_provider_names.h>
#include <yql/essentials/core/expr_nodes/yql_expr_nodes.h>
#include <yql/essentials/core/yql_expr_type_annotation.h>

namespace NYql {

using namespace NNodes;

class TPgDataSinkImpl : public TDataProviderBase {
public:
    TPgDataSinkImpl(TPgState::TPtr state)
        : State_(state)
        , TypeAnnotationTransformer_(CreatePgDataSinkTypeAnnotationTransformer(state))
        , ExecutionTransformer_(CreatePgDataSinkExecTransformer(state))
    {}

    TStringBuf GetName() const override {
        return PgProviderName;
    }

    bool CanParse(const TExprNode& node) override {
        return TypeAnnotationTransformer_->CanParse(node);
    }

    bool CanExecute(const TExprNode& node) override {
        return ExecutionTransformer_->CanExec(node);
    }

    IGraphTransformer& GetTypeAnnotationTransformer(bool instantOnly) override {
        Y_UNUSED(instantOnly);
        return *TypeAnnotationTransformer_;
    }

    IGraphTransformer& GetCallableExecutionTransformer() override {
        return *ExecutionTransformer_;
    }

    bool ValidateParameters(TExprNode& node, TExprContext& ctx, TMaybe<TString>& cluster) override {
        if (node.IsCallable(TCoDataSink::CallableName())) {
            if (!EnsureArgsCount(node, 2, ctx)) {
                return false;
            }

            if (node.Child(0)->Content() == PgProviderName) {
                if (!EnsureAtom(*node.Child(1), ctx)) {
                    return false;
                }

                if (node.Child(1)->Content().empty()) {
                    ctx.AddError(TIssue(ctx.GetPosition(node.Child(1)->Pos()), "Empty cluster name"));
                    return false;
                }

                cluster = TString(node.Child(1)->Content());
                return true;
            }
        }

        ctx.AddError(TIssue(ctx.GetPosition(node.Pos()), "Invalid Pg DataSink parameters"));
        return false;
    }

private:
    TPgState::TPtr State_;
    const THolder<TVisitorTransformerBase> TypeAnnotationTransformer_;
    const THolder<TExecTransformerBase> ExecutionTransformer_;
};

TIntrusivePtr<IDataProvider> CreatePgDataSink(TPgState::TPtr state) {
    return MakeIntrusive<TPgDataSinkImpl>(state);
}

}