/Users/deen/code/yugabyte-db/src/yb/yql/pggate/pg_doc_op.h
Line | Count | Source (jump to first uncovered line) |
1 | | //-------------------------------------------------------------------------------------------------- |
2 | | // Copyright (c) YugaByte, Inc. |
3 | | // |
4 | | // Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except |
5 | | // in compliance with the License. You may obtain a copy of the License at |
6 | | // |
7 | | // http://www.apache.org/licenses/LICENSE-2.0 |
8 | | // |
9 | | // Unless required by applicable law or agreed to in writing, software distributed under the License |
10 | | // is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
11 | | // or implied. See the License for the specific language governing permissions and limitations |
12 | | // under the License. |
13 | | //-------------------------------------------------------------------------------------------------- |
14 | | |
15 | | #ifndef YB_YQL_PGGATE_PG_DOC_OP_H_ |
16 | | #define YB_YQL_PGGATE_PG_DOC_OP_H_ |
17 | | |
18 | | #include <deque> |
19 | | |
20 | | #include <boost/optional.hpp> |
21 | | |
22 | | #include "yb/util/locks.h" |
23 | | #include "yb/client/yb_op.h" |
24 | | |
25 | | #include "yb/yql/pggate/pg_gate_fwd.h" |
26 | | #include "yb/yql/pggate/pg_op.h" |
27 | | #include "yb/yql/pggate/pg_session.h" |
28 | | |
29 | | namespace yb { |
30 | | namespace pggate { |
31 | | |
32 | | class PgTuple; |
33 | | |
34 | | YB_STRONGLY_TYPED_BOOL(RequestSent); |
35 | | |
36 | | //-------------------------------------------------------------------------------------------------- |
37 | | // PgDocResult represents a batch of rows in ONE reply from tablet servers. |
38 | | class PgDocResult { |
39 | | public: |
40 | | explicit PgDocResult(rpc::SidecarPtr&& data); |
41 | | PgDocResult(rpc::SidecarPtr&& data, std::list<int64_t>&& row_orders); |
42 | | ~PgDocResult(); |
43 | | |
44 | | PgDocResult(const PgDocResult&) = delete; |
45 | | PgDocResult& operator=(const PgDocResult&) = delete; |
46 | | |
47 | | // Get the order of the next row in this batch. |
48 | | int64_t NextRowOrder(); |
49 | | |
50 | | // End of this batch. |
51 | 54.8M | bool is_eof() const { |
52 | 54.8M | return row_count_ == 0 || row_iterator_.empty()54.0M ; |
53 | 54.8M | } |
54 | | |
55 | | // Get the postgres tuple from this batch. |
56 | | CHECKED_STATUS WritePgTuple(const std::vector<PgExpr*>& targets, PgTuple *pg_tuple, |
57 | | int64_t *row_order); |
58 | | |
59 | | // Get system columns' values from this batch. |
60 | | // Currently, we only have ybctids, but there could be more. |
61 | | CHECKED_STATUS ProcessSystemColumns(); |
62 | | |
63 | | // Update the reservoir with ybctids from this batch. |
64 | | // The update is expected to be sparse, so ybctids come as index/value pairs. |
65 | | CHECKED_STATUS ProcessSparseSystemColumns(std::string *reservoir); |
66 | | |
67 | | // Access function to ybctids value in this batch. |
68 | | // Sys columns must be processed before this function is called. |
69 | 5.63k | const vector<Slice>& ybctids() const { |
70 | 18.4E | DCHECK(syscol_processed_) << "System columns are not yet setup"; |
71 | 5.63k | return ybctids_; |
72 | 5.63k | } |
73 | | |
74 | | // Row count in this batch. |
75 | 0 | int64_t row_count() const { |
76 | 0 | return row_count_; |
77 | 0 | } |
78 | | |
79 | | private: |
80 | | // Data selected from DocDB. |
81 | | rpc::SidecarPtr data_; |
82 | | |
83 | | // Iterator on "data_" from row to row. |
84 | | Slice row_iterator_; |
85 | | |
86 | | // The row number of only this batch. |
87 | | int64_t row_count_ = 0; |
88 | | |
89 | | // The indexing order of the row in this batch. |
90 | | // These order values help to identify the row order across all batches. |
91 | | std::list<int64_t> row_orders_; |
92 | | |
93 | | // System columns. |
94 | | // - ybctids_ contains pointers to the buffers "data_". |
95 | | // - System columns must be processed before these fields have any meaning. |
96 | | vector<Slice> ybctids_; |
97 | | bool syscol_processed_ = false; |
98 | | }; |
99 | | |
100 | | //-------------------------------------------------------------------------------------------------- |
101 | | // Doc operation API |
102 | | // Classes |
103 | | // - PgDocOp: Shared functionalities among all ops, mostly just RPC calls to tablet servers. |
104 | | // - PgDocReadOp: Definition for data & method members to be used in READ operation. |
105 | | // - PgDocWriteOp: Definition for data & method members to be used in WRITE operation. |
106 | | // - PgDocResult: Definition data holder before they are passed to Postgres layer. |
107 | | // |
108 | | // Processing Steps |
109 | | // (1) Collecting Data: |
110 | | // PgGate collects data from Posgres and write to a "PgDocOp::Template". |
111 | | // |
112 | | // (2) Create operators: |
113 | | // When no optimization is applied, the "template_op" is executed as is. When an optimization |
114 | | // is chosen, PgDocOp will clone the template to populate operators and kept them in vector |
115 | | // "pgsql_ops_". When an op executes arguments, it sends request and reads replies from servers. |
116 | | // |
117 | | // * Vector "pgsql_ops_" is of fixed size for the entire execution, and its contents (YBPgsqlOp |
118 | | // shared_ptrs) also remain for the entire execution. |
119 | | // * There is a LIMIT on how many pgsql-op can be cloned. If the number of requests / arguments |
120 | | // are higher than the LIMIT, some requests will have to wait in queue until the execution |
121 | | // of precedent arguments are completed. |
122 | | // * After an argument input is executed, its associated YBPgsqlOp will be reused to execute |
123 | | // a new set of arguments. We don't clone new ones for new arguments. |
124 | | // * When a YBPgsqlOp is reused, its YBPgsqlOp::ProtobufRequest will be updated appropriately |
125 | | // with new arguments. |
126 | | // * NOTE: Some operators in "pgsql_ops_" might not be active (no arguments) at a given time |
127 | | // of execution. For example, some ops might complete their execution while others have |
128 | | // paging state and are sent again to table server. |
129 | | // |
130 | | // (3) SendRequest: |
131 | | // PgSession API requires contiguous array of operators. For this reason, before sending the |
132 | | // pgsql_ops_ is soreted to place active ops first, and all inactive ops are place at the end. |
133 | | // For example, |
134 | | // PgSession::RunAsync(pgsql_ops_.data(), active_op_count) |
135 | | // |
136 | | // (4) ReadResponse: |
137 | | // Response are written to a local cache PgDocResult. |
138 | | // |
139 | | // This API has several sets of methods and attributes for different purposes. |
140 | | // (1) Build request. |
141 | | // This section collect information and data from PgGate API. |
142 | | // * Attributes |
143 | | // - relation_id_: Table to be operated on. |
144 | | // - template_op_ of type YBPgsqlReadOp and YBPgsqlWriteOp. |
145 | | // This object contains statement descriptions and expression values from users. |
146 | | // All user-provided arguments are kept in this attributes. |
147 | | // * Methods |
148 | | // - Class constructors. |
149 | | // |
150 | | // (2) Constructing protobuf request. |
151 | | // This section populates protobuf requests using the collected information in "template_op_". |
152 | | // - Without optimization, the protobuf request in "template_op_" will be used . |
153 | | // - With parallel optimization, multiple protobufs are constructed by cloning template into many |
154 | | // operators. How the execution are subdivided is depending on the parallelism method. |
155 | | // NOTE Whenever we support PREPARE(stmt), we'd stop processing at after this step for PREPARE. |
156 | | // |
157 | | // * Attributes |
158 | | // - YBPgsqlOp pgsql_ops_: Contains all protobuf requests to be sent to tablet servers. |
159 | | // * Methods |
160 | | // - When there isn't any optimization, template_op_ is used. |
161 | | // pgsql_ops_[0] = template_op_ |
162 | | // - CreateRequests() |
163 | | // - ClonePgsqlOps() Clone template_op_ into one or more ops. |
164 | | // - PopulateParallelSelectOps() Parallel processing of aggregate requests or requests with |
165 | | // WHERE expressions filtering rows in DocDB. |
166 | | // The same requests are constructed for each tablet server. |
167 | | // - PopulateNextHashPermutationOps() Parallel processing SELECT by hash conditions. |
168 | | // Hash permutations will be group into different request based on their hash_codes. |
169 | | // - PopulateDmlByYbctidOps() Parallel processing SELECT by ybctid values. |
170 | | // Ybctid values will be group into different request based on their hash_codes. |
171 | | // This function is a bit different from other formulating function because it is used for an |
172 | | // internal request within PgGate. Other populate functions are used for external requests |
173 | | // from Postgres layer via PgGate API. |
174 | | // |
175 | | // (3) Execution |
176 | | // This section exchanges RPC calls with tablet servers. |
177 | | // * Attributes |
178 | | // - active_op_counts_: Number of active operators in vector "pgsql_ops_". |
179 | | // Exec/active op range = pgsql_ops_[0, active_op_count_) |
180 | | // Inactive op range = pgsql_ops_[active_op_count_, total_count) |
181 | | // The vector pgsql_ops_ is fixed sized, can have inactive operators as operators are not |
182 | | // completing execution at the same time. |
183 | | // * Methods |
184 | | // - ExecuteInit() |
185 | | // - Execute() Driver for all RPC related effort. |
186 | | // - SendRequest() Send request for active operators to tablet server using YBPgsqlOp. |
187 | | // RunAsync(pgsql_ops_.data(), active_op_count_) |
188 | | // - ProcessResponse() Get response from tablet server using YBPgsqlOp. |
189 | | // - MoveInactiveOpsOutside() Sort pgsql_ops_ to move inactive operators outside of exec range. |
190 | | // |
191 | | // (4) Return result |
192 | | // This section return result via PgGate API to postgres. |
193 | | // * Attributes |
194 | | // - Objects of class PgDocResult |
195 | | // - rows_affected_count_: Number of rows that was operated by this doc_op. |
196 | | // * Methods |
197 | | // - GetResult() |
198 | | // - GetRowsAffectedCount() |
199 | | // |
200 | | // TODO(dmitry / neil) Allow sending active requests and receive their response one at a time. |
201 | | // |
202 | | // To process data in parallel, the operators must be able to run independently from one another. |
203 | | // However, currently operators are executed in batches and together even though they belong to |
204 | | // different partitions and interact with different tablet servers. |
205 | | //-------------------------------------------------------------------------------------------------- |
206 | | |
207 | | class PgDocOp : public std::enable_shared_from_this<PgDocOp> { |
208 | | public: |
209 | | // Public types. |
210 | | typedef std::shared_ptr<PgDocOp> SharedPtr; |
211 | | typedef std::shared_ptr<const PgDocOp> SharedPtrConst; |
212 | | |
213 | | typedef std::unique_ptr<PgDocOp> UniPtr; |
214 | | typedef std::unique_ptr<const PgDocOp> UniPtrConst; |
215 | | |
216 | | // Constructors & Destructors. |
217 | | PgDocOp(const PgSession::ScopedRefPtr& pg_session, PgTable* table); |
218 | | virtual ~PgDocOp(); |
219 | | |
220 | | // Initialize doc operator. |
221 | | virtual CHECKED_STATUS ExecuteInit(const PgExecParameters *exec_params); |
222 | | |
223 | | const PgExecParameters& ExecParameters() const; |
224 | | |
225 | | // Execute the op. Return true if the request has been sent and is awaiting the result. |
226 | | virtual Result<RequestSent> Execute(bool force_non_bufferable = false); |
227 | | |
228 | | // Instruct this doc_op to abandon execution and querying data by setting end_of_data_ to 'true'. |
229 | | // - This op will not send request to tablet server. |
230 | | // - This op will return empty result-set when being requested for data. |
231 | 492 | void AbandonExecution() { |
232 | 492 | end_of_data_ = true; |
233 | 492 | } |
234 | | |
235 | | // Get the result of the op. No rows will be added to rowsets in case end of data reached. |
236 | | CHECKED_STATUS GetResult(std::list<PgDocResult> *rowsets); |
237 | | Result<int32_t> GetRowsAffectedCount() const; |
238 | | |
239 | | // This operation is requested internally within PgGate, and that request does not go through |
240 | | // all the steps as other operation from Postgres thru PgDocOp. This is used to create requests |
241 | | // for the following select. |
242 | | // SELECT ... FROM <table> WHERE ybctid IN (SELECT base_ybctids from INDEX) |
243 | | // After ybctids are queried from INDEX, PgGate will call "PopulateDmlByYbctidOps" to create |
244 | | // operators to fetch rows whose rowids equal queried ybctids. |
245 | | CHECKED_STATUS PopulateDmlByYbctidOps(const std::vector<Slice>& ybctids); |
246 | | |
247 | 1.65M | bool has_out_param_backfill_spec() { |
248 | 1.65M | return !out_param_backfill_spec_.empty(); |
249 | 1.65M | } |
250 | | |
251 | 2.25k | const char* out_param_backfill_spec() { |
252 | 2.25k | return out_param_backfill_spec_.c_str(); |
253 | 2.25k | } |
254 | | |
255 | 0 | bool end_of_data() const { |
256 | 0 | return end_of_data_; |
257 | 0 | } |
258 | | |
259 | | virtual bool IsWrite() const = 0; |
260 | | |
261 | | CHECKED_STATUS CreateRequests(); |
262 | | |
263 | | protected: |
264 | | uint64_t& GetReadTime(); |
265 | | |
266 | | // Populate Protobuf requests using the collected informtion for this DocDB operator. |
267 | | virtual Result<bool> DoCreateRequests() = 0; |
268 | | |
269 | | virtual CHECKED_STATUS DoPopulateDmlByYbctidOps(const std::vector<Slice>& ybctids) = 0; |
270 | | |
271 | | // Create operators. |
272 | | // - Each operator is used for one request. |
273 | | // - When parallelism by partition is applied, each operator is associated with one partition, |
274 | | // and each operator has a batch of arguments that belong to that partition. |
275 | | // * The higher the number of partition_count, the higher the parallelism level. |
276 | | // * If (partition_count == 1), only one operator is needed for the entire partition range. |
277 | | // * If (partition_count > 1), each operator is used for a specific partition range. |
278 | | // * This optimization is used by |
279 | | // PopulateDmlByYbctidOps() |
280 | | // PopulateParallelSelectOps() |
281 | | // - When parallelism by arguments is applied, each operator has only one argument. |
282 | | // When tablet server will run the requests in parallel as it assigned one thread per request. |
283 | | // PopulateNextHashPermutationOps() |
284 | | CHECKED_STATUS ClonePgsqlOps(size_t op_count); |
285 | | |
286 | | // Only active operators are kept in the active range [0, active_op_count_) |
287 | | // - Not execute operators that are outside of range [0, active_op_count_). |
288 | | // - Sort the operators in "pgsql_ops_" to move "inactive" operators to the end of the list. |
289 | | void MoveInactiveOpsOutside(); |
290 | | |
291 | | // Clone READ or WRITE "template_op_" into new operators. |
292 | | virtual PgsqlOpPtr CloneFromTemplate() = 0; |
293 | | |
294 | | // Process the result set in server response. |
295 | | Result<std::list<PgDocResult>> ProcessResponseResult(); |
296 | | |
297 | | void SetReadTime(); |
298 | | |
299 | | private: |
300 | | CHECKED_STATUS SendRequest(bool force_non_bufferable); |
301 | | |
302 | | virtual CHECKED_STATUS SendRequestImpl(bool force_non_bufferable); |
303 | | |
304 | | Result<std::list<PgDocResult>> ProcessResponse(const Status& exec_status); |
305 | | |
306 | | virtual Result<std::list<PgDocResult>> ProcessResponseImpl() = 0; |
307 | | |
308 | | CHECKED_STATUS CompleteRequests(); |
309 | | |
310 | | //----------------------------------- Data Members ----------------------------------------------- |
311 | | protected: |
312 | | // Session control. |
313 | | PgSession::ScopedRefPtr pg_session_; |
314 | | |
315 | | // Operation time. This time is set at the start and must stay the same for the lifetime of the |
316 | | // operation to ensure that it is operating on one snapshot. |
317 | | uint64_t read_time_ = 0; |
318 | | |
319 | | // Target table. |
320 | | PgTable& table_; |
321 | | |
322 | | // Exec control parameters. |
323 | | PgExecParameters exec_params_; |
324 | | |
325 | | // Suppress sending new request after processing response. |
326 | | // Next request will be sent in case upper level will ask for additional data. |
327 | | bool suppress_next_result_prefetching_ = false; |
328 | | |
329 | | // Populated protobuf request. |
330 | | std::vector<PgsqlOpPtr> pgsql_ops_; |
331 | | |
332 | | // Number of active operators in the pgsql_ops_ list. |
333 | | size_t active_op_count_ = 0; |
334 | | |
335 | | // Indicator for completing all request populations. |
336 | | bool request_population_completed_ = false; |
337 | | |
338 | | // If true, all data for each batch must be collected before PgGate gets the reply. |
339 | | // NOTE: |
340 | | // - Currently, PgSession's default behavior is to get all responses in a batch together. |
341 | | // - We set this flag only to prevent future optimization where requests & their responses to |
342 | | // and from different tablet servers are sent and received independently. That optimization |
343 | | // should only be done when "wait_for_batch_completion_ == false" |
344 | | bool wait_for_batch_completion_ = true; |
345 | | |
346 | | // Future object to fetch a response from DocDB after sending a request. |
347 | | // Object's valid() method returns false in case no request is sent |
348 | | // or sent request was buffered by the session. |
349 | | // Only one RunAsync() can be called to sent to DocDB at a time. |
350 | | PerformFuture response_; |
351 | | |
352 | | // Executed row count. |
353 | | int32_t rows_affected_count_ = 0; |
354 | | |
355 | | // Whether all requested data by the statement has been received or there's a run-time error. |
356 | | bool end_of_data_ = false; |
357 | | |
358 | | // The order number of each request when batching arguments. |
359 | | // Currently, this is used for query by YBCTID. |
360 | | // - Each pgsql_op has a batch of ybctids selected from INDEX. |
361 | | // - The order of resulting rows should match with the order of queried ybctids. |
362 | | // - Example: |
363 | | // Suppose we got from INDEX table |
364 | | // { ybctid_1, ybctid_2, ybctid_3, ybctid_4, ybctid_5, ybctid_6, ybctid_7 } |
365 | | // |
366 | | // Now pgsql_op are constructed as the following, one op per partition. |
367 | | // pgsql_op <partition 1> (ybctid_1, ybctid_3, ybctid_4) |
368 | | // pgsql_op <partition 2> (ybctid_2, ybctid_6) |
369 | | // pgsql_op <partition 2> (ybctid_5, ybctid_7) |
370 | | // |
371 | | // These respective ybctids are stored in batch_ybctid_ also. |
372 | | // In other words, |
373 | | // batch_ybctid_[partition 1] contains (ybctid_1, ybctid_3, ybctid_4) |
374 | | // batch_ybctid_[partition 2] contains (ybctid_2, ybctid_6) |
375 | | // batch_ybctid_[partition 3] contains (ybctid_5, ybctid_7) |
376 | | // |
377 | | // After getting the rows of data from pgsql, the rows must be then ordered from 1 thru 7. |
378 | | // To do so, for each pgsql_op we kept an array of orders, batch_row_orders_. |
379 | | // For the above pgsql_ops_, the orders would be cached as the following. |
380 | | // vector orders { partition 1: list ( 1, 3, 4 ), |
381 | | // partition 2: list ( 2, 6 ), |
382 | | // partition 3: list ( 5, 7 ) } |
383 | | // |
384 | | // When the "pgsql_ops_" elements are sorted and swapped order, the "batch_row_orders_" |
385 | | // must be swaped also. |
386 | | // std::swap ( pgsql_ops_[1], pgsql_ops_[3]) |
387 | | // std::swap ( batch_row_orders_[1], batch_row_orders_[3] ) |
388 | | std::vector<std::list<int64_t>> batch_row_orders_; |
389 | | |
390 | | // This counter is used to maintain the row order when the operator sends requests in parallel |
391 | | // by partition. Currently only query by YBCTID uses this variable. |
392 | | int64_t batch_row_ordering_counter_ = 0; |
393 | | |
394 | | // Parallelism level. |
395 | | // - This is the maximum number of read/write requests being sent to servers at one time. |
396 | | // - When it is 1, there's no optimization. Available requests is executed one at a time. |
397 | | size_t parallelism_level_ = 1; |
398 | | |
399 | | // Output parameter of the execution. |
400 | | string out_param_backfill_spec_; |
401 | | |
402 | | private: |
403 | | // Result set either from selected or returned targets is cached in a list of strings. |
404 | | // Querying state variables. |
405 | | Status exec_status_ = Status::OK(); |
406 | | }; |
407 | | |
408 | | //-------------------------------------------------------------------------------------------------- |
409 | | |
410 | | class PgDocReadOp : public PgDocOp { |
411 | | public: |
412 | | // Public types. |
413 | | typedef std::shared_ptr<PgDocReadOp> SharedPtr; |
414 | | typedef std::shared_ptr<const PgDocReadOp> SharedPtrConst; |
415 | | |
416 | | typedef std::unique_ptr<PgDocReadOp> UniPtr; |
417 | | typedef std::unique_ptr<const PgDocReadOp> UniPtrConst; |
418 | | |
419 | | // Constructors & Destructors. |
420 | | PgDocReadOp( |
421 | | const PgSession::ScopedRefPtr& pg_session, PgTable* table, |
422 | | PgsqlReadOpPtr read_op); |
423 | | |
424 | | CHECKED_STATUS ExecuteInit(const PgExecParameters *exec_params) override; |
425 | | |
426 | | // Row sampler collects number of live and dead rows it sees. |
427 | | CHECKED_STATUS GetEstimatedRowCount(double *liverows, double *deadrows); |
428 | | |
429 | 0 | bool IsWrite() const override { |
430 | 0 | return false; |
431 | 0 | } |
432 | | |
433 | | CHECKED_STATUS DoPopulateDmlByYbctidOps(const std::vector<Slice>& ybctids) override; |
434 | | |
435 | | private: |
436 | | // Create protobuf requests using template_op_. |
437 | | Result<bool> DoCreateRequests() override; |
438 | | |
439 | | // Create operators by partition. |
440 | | // - Optimization for statement |
441 | | // SELECT xxx FROM <table> WHERE ybctid IN (SELECT ybctid FROM INDEX) |
442 | | // - After being queried from inner select, ybctids are used for populate request for outer query. |
443 | | CHECKED_STATUS InitializeYbctidOperators(); |
444 | | |
445 | | // Create operators by partition arguments. |
446 | | // - Optimization for statement: |
447 | | // SELECT ... WHERE <hash-columns> IN <value-lists> |
448 | | // - If partition column binds are defined, partition_column_values field of each operation |
449 | | // is set to be the next permutation. |
450 | | // - When an operator is assigned a hash permutation, it is marked as active to be executed. |
451 | | // - When an operator completes the execution, it is marked as inactive and available for the |
452 | | // exection of the next hash permutation. |
453 | | Result<bool> PopulateNextHashPermutationOps(); |
454 | | CHECKED_STATUS InitializeHashPermutationStates(); |
455 | | |
456 | | // Create operators by partitions. |
457 | | // - Optimization for aggregating or filtering requests. |
458 | | Result<bool> PopulateParallelSelectOps(); |
459 | | |
460 | | // Create one sampling operator per partition and arrange their execution in random order |
461 | | Result<bool> PopulateSamplingOps(); |
462 | | |
463 | | // Set partition boundaries to a given partition. |
464 | | CHECKED_STATUS SetScanPartitionBoundary(); |
465 | | |
466 | | // Process response from DocDB. |
467 | | Result<std::list<PgDocResult>> ProcessResponseImpl() override; |
468 | | |
469 | | // Process response read state from DocDB. |
470 | | CHECKED_STATUS ProcessResponseReadStates(); |
471 | | |
472 | | // Reset pgsql operators before reusing them with new arguments / inputs from Postgres. |
473 | | CHECKED_STATUS ResetInactivePgsqlOps(); |
474 | | |
475 | | // Analyze options and pick the appropriate prefetch limit. |
476 | | void SetRequestPrefetchLimit(); |
477 | | |
478 | | // Set the backfill_spec field of our read request. |
479 | | void SetBackfillSpec(); |
480 | | |
481 | | // Set the row_mark_type field of our read request based on our exec control parameter. |
482 | | void SetRowMark(); |
483 | | |
484 | | // Set the read_time for our read request based on our exec control parameter. |
485 | | void SetReadTime(); |
486 | | |
487 | | // Clone the template into actual requests to be sent to server. |
488 | 2.95M | PgsqlOpPtr CloneFromTemplate() override { |
489 | 2.95M | return read_op_->DeepCopy(); |
490 | 2.95M | } |
491 | | |
492 | | // Get the read_req for a specific operation index from pgsql_ops_. |
493 | | PgsqlReadRequestPB& GetReadReq(size_t op_index); |
494 | | |
495 | | // Re-format the request when connecting to older server during rolling upgrade. |
496 | | void FormulateRequestForRollingUpgrade(PgsqlReadRequestPB *read_req); |
497 | | |
498 | | //----------------------------------- Data Members ----------------------------------------------- |
499 | | |
500 | | // Template operation, used to fill in pgsql_ops_ by either assigning or cloning. |
501 | | PgsqlReadOpPtr read_op_; |
502 | | |
503 | | // While sampling is in progress, number of scanned row is accumulated in this variable. |
504 | | // After completion the value is extrapolated to account for not scanned partitions and estimate |
505 | | // total number of rows in the table. |
506 | | double sample_rows_ = 0; |
507 | | |
508 | | // Used internally for PopulateNextHashPermutationOps to keep track of which permutation should |
509 | | // be used to construct the next read_op. |
510 | | // Is valid as long as request_population_completed_ is false. |
511 | | // |
512 | | // Example: |
513 | | // For a query clause "h1 = 1 AND h2 IN (2,3) AND h3 IN (4,5,6) AND h4 = 7", |
514 | | // there are 1*2*3*1 = 6 possible permutation. |
515 | | // As such, this field will take on values 0 through 5. |
516 | | int total_permutation_count_ = 0; |
517 | | int next_permutation_idx_ = 0; |
518 | | |
519 | | // Used internally for PopulateNextHashPermutationOps to holds all partition expressions. |
520 | | // Elements correspond to a hash columns, in the same order as they were defined |
521 | | // in CREATE TABLE statement. |
522 | | // This is somewhat similar to what hash_values_options_ in CQL is used for. |
523 | | // |
524 | | // Example: |
525 | | // For a query clause "h1 = 1 AND h2 IN (2,3) AND h3 IN (4,5,6) AND h4 = 7", |
526 | | // this will be initialized to [[1], [2, 3], [4, 5, 6], [7]] |
527 | | std::vector<std::vector<const PgsqlExpressionPB*>> partition_exprs_; |
528 | | }; |
529 | | |
530 | | //-------------------------------------------------------------------------------------------------- |
531 | | |
532 | | class PgDocWriteOp : public PgDocOp { |
533 | | public: |
534 | | // Public types. |
535 | | typedef std::shared_ptr<PgDocWriteOp> SharedPtr; |
536 | | typedef std::shared_ptr<const PgDocWriteOp> SharedPtrConst; |
537 | | |
538 | | typedef std::unique_ptr<PgDocWriteOp> UniPtr; |
539 | | typedef std::unique_ptr<const PgDocWriteOp> UniPtrConst; |
540 | | |
541 | | // Constructors & Destructors. |
542 | | PgDocWriteOp(const PgSession::ScopedRefPtr& pg_session, |
543 | | PgTable* table, |
544 | | PgsqlWriteOpPtr write_op); |
545 | | |
546 | | // Set write time. |
547 | | void SetWriteTime(const HybridTime& write_time); |
548 | | |
549 | 0 | bool IsWrite() const override { |
550 | 0 | return true; |
551 | 0 | } |
552 | | |
553 | | private: |
554 | | // Process response implementation. |
555 | | Result<std::list<PgDocResult>> ProcessResponseImpl() override; |
556 | | |
557 | | // Create protobuf requests using template_op (write_op). |
558 | | Result<bool> DoCreateRequests() override; |
559 | | |
560 | | // For write ops, we are not yet batching ybctid from index query. |
561 | | // TODO(neil) This function will be implemented when we push down sub-query inside WRITE ops to |
562 | | // the proxy layer. There's many scenarios where this optimization can be done. |
563 | 0 | CHECKED_STATUS DoPopulateDmlByYbctidOps(const std::vector<Slice>& ybctids) override { |
564 | 0 | LOG(FATAL) << "Not yet implemented"; |
565 | 0 | return Status::OK(); |
566 | 0 | } |
567 | | |
568 | | // Get WRITE operator for a specific operator index in pgsql_ops_. |
569 | | PgsqlWriteRequestPB& GetWriteOp(int op_index); |
570 | | |
571 | | // Clone user data from template to actual protobuf requests. |
572 | 0 | PgsqlOpPtr CloneFromTemplate() override { |
573 | 0 | return write_op_->DeepCopy(); |
574 | 0 | } |
575 | | |
576 | | //----------------------------------- Data Members ----------------------------------------------- |
577 | | // Template operation all write ops. |
578 | | PgsqlWriteOpPtr write_op_; |
579 | | }; |
580 | | |
581 | | } // namespace pggate |
582 | | } // namespace yb |
583 | | |
584 | | #endif // YB_YQL_PGGATE_PG_DOC_OP_H_ |