Line data Source code
1 : /**
2 : Copyright (c) 2017-2022 Roman Katuntsev <sbkarr@stappler.org>
3 : Copyright (c) 2023-2024 Stappler LLC <admin@stappler.dev>
4 :
5 : Permission is hereby granted, free of charge, to any person obtaining a copy
6 : of this software and associated documentation files (the "Software"), to deal
7 : in the Software without restriction, including without limitation the rights
8 : to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 : copies of the Software, and to permit persons to whom the Software is
10 : furnished to do so, subject to the following conditions:
11 :
12 : The above copyright notice and this permission notice shall be included in
13 : all copies or substantial portions of the Software.
14 :
15 : THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 : IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 : FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 : AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 : LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 : OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 : THE SOFTWARE.
22 : **/
23 :
24 : #include "SPPqHandle.h"
25 :
26 : namespace STAPPLER_VERSIONIZED stappler::db::pq {
27 :
28 : struct ExecParamData {
29 : std::array<const char *, 64> values;
30 : std::array<int, 64> sizes;
31 : std::array<int, 64> formats;
32 :
33 : Vector<const char *> valuesVec;
34 : Vector<int> sizesVec;
35 : Vector<int> formatsVec;
36 :
37 : const char *const * paramValues = nullptr;
38 : const int *paramLengths = nullptr;
39 : const int *paramFormats = nullptr;
40 :
41 3900 : ExecParamData(const db::sql::SqlQuery &query) {
42 3900 : auto queryInterface = static_cast<PgQueryInterface *>(query.getInterface());
43 :
44 3900 : auto size = queryInterface->params.size();
45 :
46 3900 : if (size > 64) {
47 0 : valuesVec.reserve(size);
48 0 : sizesVec.reserve(size);
49 0 : formatsVec.reserve(size);
50 :
51 0 : for (size_t i = 0; i < size; ++ i) {
52 0 : const Bytes &d = queryInterface->params.at(i);
53 0 : bool bin = queryInterface->binary.at(i);
54 0 : valuesVec.emplace_back((const char *)d.data());
55 0 : sizesVec.emplace_back(int(d.size()));
56 0 : formatsVec.emplace_back(bin);
57 : }
58 :
59 0 : paramValues = valuesVec.data();
60 0 : paramLengths = sizesVec.data();
61 0 : paramFormats = formatsVec.data();
62 : } else {
63 10525 : for (size_t i = 0; i < size; ++ i) {
64 6625 : const Bytes &d = queryInterface->params.at(i);
65 6625 : bool bin = queryInterface->binary.at(i);
66 6625 : values[i] = (const char *)d.data();
67 6625 : sizes[i] = int(d.size());
68 6625 : formats[i] = bin;
69 : }
70 :
71 3900 : paramValues = values.data();
72 3900 : paramLengths = sizes.data();
73 3900 : paramFormats = formats.data();
74 : }
75 3900 : }
76 : };
77 :
78 0 : SPUNUSED static String pg_numeric_to_string(BytesViewNetwork r) {
79 : using NumericDigit = int16_t;
80 : static constexpr auto DEC_DIGITS = 4;
81 : static constexpr auto NUMERIC_NEG = 0x4000;
82 :
83 : int d;
84 : NumericDigit dig;
85 : NumericDigit d1;
86 :
87 0 : uint16_t ndigits = uint16_t(r.readUnsigned16());
88 0 : int16_t weight = int16_t(r.readUnsigned16());
89 0 : int16_t sign = int16_t(r.readUnsigned16());
90 0 : int16_t dscale = int16_t(r.readUnsigned16());
91 :
92 0 : Vector<int16_t> digits;
93 0 : for (uint16_t i = 0; i < ndigits && !r.empty(); i++) {
94 0 : digits.emplace_back(r.readUnsigned16());
95 : }
96 :
97 0 : int i = (weight + 1) * DEC_DIGITS;
98 0 : if (i <= 0) {
99 0 : i = 1;
100 : }
101 :
102 0 : String str; str.reserve(i + dscale + DEC_DIGITS + 2);
103 :
104 0 : if (sign == NUMERIC_NEG) { str.push_back('-'); }
105 0 : if (weight < 0) {
106 0 : d = weight + 1;
107 0 : str.push_back('0');
108 : } else {
109 0 : for (d = 0; d <= weight; d++) {
110 0 : dig = (d < ndigits) ? digits[d] : 0;
111 :
112 0 : bool putit = (d > 0);
113 0 : d1 = dig / 1000;
114 0 : dig -= d1 * 1000;
115 0 : putit |= (d1 > 0);
116 0 : if (putit) { str.push_back(d1 + '0'); }
117 0 : d1 = dig / 100;
118 0 : dig -= d1 * 100;
119 0 : putit |= (d1 > 0);
120 0 : if (putit) { str.push_back(d1 + '0'); }
121 0 : d1 = dig / 10;
122 0 : dig -= d1 * 10;
123 0 : putit |= (d1 > 0);
124 0 : if (putit) { str.push_back(d1 + '0'); }
125 0 : str.push_back(dig + '0');
126 : }
127 : }
128 :
129 0 : if (dscale > 0) {
130 0 : str.push_back('.');
131 0 : for (i = 0; i < dscale; d++, i += DEC_DIGITS) {
132 0 : dig = (d >= 0 && d < ndigits) ? digits[d] : 0;
133 :
134 0 : d1 = dig / 1000;
135 0 : dig -= d1 * 1000;
136 0 : str.push_back(d1 + '0');
137 0 : d1 = dig / 100;
138 0 : dig -= d1 * 100;
139 0 : str.push_back(d1 + '0');
140 0 : d1 = dig / 10;
141 0 : dig -= d1 * 10;
142 0 : str.push_back(d1 + '0');
143 0 : str.push_back(dig + '0');
144 : }
145 : }
146 :
147 0 : return str;
148 0 : }
149 3875 : PgQueryInterface::PgQueryInterface(const sql::Driver *d, const sql::QueryStorageHandle *s)
150 3875 : : driver(d), storage(s) { }
151 :
152 50 : size_t PgQueryInterface::push(String &&val) {
153 50 : params.emplace_back(Bytes());
154 50 : params.back().assign_strong((uint8_t *)val.data(), val.size() + 1);
155 50 : binary.emplace_back(false);
156 50 : return params.size();
157 : }
158 :
159 750 : size_t PgQueryInterface::push(const StringView &val) {
160 750 : params.emplace_back(Bytes());
161 750 : params.back().assign_strong((uint8_t *)val.data(), val.size() + 1);
162 750 : binary.emplace_back(false);
163 750 : return params.size();
164 : }
165 :
166 0 : size_t PgQueryInterface::push(Bytes &&val) {
167 0 : params.emplace_back(std::move(val));
168 0 : binary.emplace_back(true);
169 0 : return params.size();
170 : }
171 10000 : size_t PgQueryInterface::push(StringStream &query, const Value &val, bool force, bool compress) {
172 10000 : if (!force || val.getType() == Value::Type::EMPTY) {
173 8450 : switch (val.getType()) {
174 0 : case Value::Type::EMPTY:
175 0 : query << "NULL";
176 0 : break;
177 825 : case Value::Type::BOOLEAN:
178 825 : if (val.asBool()) {
179 400 : query << "TRUE";
180 : } else {
181 425 : query << "FALSE";
182 : }
183 825 : break;
184 2600 : case Value::Type::INTEGER:
185 2600 : query << val.asInteger();
186 2600 : break;
187 750 : case Value::Type::DOUBLE:
188 750 : if (std::isnan(val.asDouble())) {
189 0 : query << "NaN";
190 750 : } else if (val.asDouble() == std::numeric_limits<double>::infinity()) {
191 0 : query << "-Infinity";
192 750 : } else if (-val.asDouble() == std::numeric_limits<double>::infinity()) {
193 0 : query << "Infinity";
194 : } else {
195 750 : query << std::setprecision(std::numeric_limits<double>::max_digits10 + 1) << val.asDouble();
196 : }
197 750 : break;
198 3525 : case Value::Type::CHARSTRING:
199 3525 : params.emplace_back(Bytes());
200 3525 : params.back().assign_strong((uint8_t *)val.getString().data(), val.size() + 1);
201 3525 : binary.emplace_back(false);
202 3525 : query << "$" << params.size() << "::text";
203 3525 : break;
204 750 : case Value::Type::BYTESTRING:
205 750 : params.emplace_back(val.asBytes());
206 750 : binary.emplace_back(true);
207 750 : query << "$" << params.size() << "::bytea";
208 750 : break;
209 0 : case Value::Type::ARRAY:
210 : case Value::Type::DICTIONARY:
211 0 : params.emplace_back(data::write<Interface>(val, EncodeFormat(EncodeFormat::Cbor,
212 0 : compress ? EncodeFormat::LZ4HCCompression : EncodeFormat::DefaultCompress)));
213 0 : binary.emplace_back(true);
214 0 : query << "$" << params.size() << "::bytea";
215 0 : break;
216 0 : default: break;
217 : }
218 : } else {
219 1550 : params.emplace_back(data::write<Interface>(val, EncodeFormat(EncodeFormat::Cbor,
220 1550 : compress ? EncodeFormat::LZ4HCCompression : EncodeFormat::DefaultCompress)));
221 1550 : binary.emplace_back(true);
222 1550 : query << "$" << params.size() << "::bytea";
223 : }
224 10000 : return params.size();
225 : }
226 :
227 1625 : void PgQueryInterface::bindInt(db::Binder &, StringStream &query, int64_t val) {
228 1625 : query << val;
229 1625 : }
230 :
231 900 : void PgQueryInterface::bindUInt(db::Binder &, StringStream &query, uint64_t val) {
232 900 : query << val;
233 900 : }
234 :
235 0 : void PgQueryInterface::bindDouble(db::Binder &, StringStream &query, double val) {
236 0 : query << std::setprecision(std::numeric_limits<double>::max_digits10 + 1) << val;
237 0 : }
238 :
239 0 : void PgQueryInterface::bindString(db::Binder &, StringStream &query, const String &val) {
240 0 : if (auto num = push(String(val))) {
241 0 : query << "$" << num << "::text";
242 : }
243 0 : }
244 :
245 0 : void PgQueryInterface::bindMoveString(db::Binder &, StringStream &query, String &&val) {
246 0 : if (auto num = push(std::move(val))) {
247 0 : query << "$" << num << "::text";
248 : }
249 0 : }
250 :
251 0 : void PgQueryInterface::bindStringView(db::Binder &, StringStream &query, const StringView &val) {
252 0 : if (auto num = push(val)) {
253 0 : query << "$" << num << "::text";
254 : }
255 0 : }
256 :
257 0 : void PgQueryInterface::bindBytes(db::Binder &, StringStream &query, const Bytes &val) {
258 0 : if (auto num = push(Bytes(val))) {
259 0 : query << "$" << num << "::bytea";
260 : }
261 0 : }
262 :
263 0 : void PgQueryInterface::bindMoveBytes(db::Binder &, StringStream &query, Bytes &&val) {
264 0 : if (auto num = push(std::move(val))) {
265 0 : query << "$" << num << "::bytea";
266 : }
267 0 : }
268 :
269 0 : void PgQueryInterface::bindCoderSource(db::Binder &, StringStream &query, const stappler::CoderSource &val) {
270 0 : if (auto num = push(Bytes(val.data(), val.data() + val.size()))) {
271 0 : query << "$" << num << "::bytea";
272 : }
273 0 : }
274 :
275 25 : void PgQueryInterface::bindValue(db::Binder &, StringStream &query, const Value &val) {
276 25 : push(query, val, false);
277 25 : }
278 :
279 9975 : void PgQueryInterface::bindDataField(db::Binder &, StringStream &query, const db::Binder::DataField &f) {
280 9975 : if (f.field && f.field->getType() == db::Type::Custom) {
281 2250 : auto custom = f.field->getSlot<db::FieldCustom>();
282 2250 : if (auto info = driver->getCustomFieldInfo(custom->getDriverTypeName())) {
283 2250 : if (!info->writeToStorage(*custom, *this, query, f.data)) {
284 0 : query << "NULL";
285 : }
286 : } else {
287 0 : query << "NULL";
288 : }
289 : } else {
290 7725 : push(query, f.data, f.force, f.compress);
291 : }
292 9975 : }
293 :
294 0 : void PgQueryInterface::bindTypeString(db::Binder &, StringStream &query, const db::Binder::TypeString &type) {
295 0 : if (auto num = push(type.str)) {
296 0 : query << "$" << num << "::" << type.type;
297 : }
298 0 : }
299 :
300 750 : void PgQueryInterface::bindFullText(db::Binder &, StringStream &query, const db::Binder::FullTextField &d) {
301 750 : if (d.data.empty()) {
302 0 : query << "NULL";
303 : } else {
304 750 : auto slot = d.field->getSlot<FieldFullTextView>();
305 750 : auto str = slot->searchConfiguration->encodeSearchVectorPostgres(d.data);
306 750 : auto dataIdx = push(str);
307 750 : query << " $" << dataIdx << "::tsvector";
308 750 : }
309 750 : }
310 :
311 50 : void PgQueryInterface::bindFullTextFrom(db::Binder &, StringStream &query, const db::Binder::FullTextFrom &) {
312 :
313 50 : }
314 :
315 50 : void PgQueryInterface::bindFullTextRank(db::Binder &, StringStream &query, const db::Binder::FullTextRank &d) {
316 50 : auto slot = d.field->getSlot<FieldFullTextView>();
317 50 : query << " ts_rank(" << d.scheme << ".\"" << d.field->getName() << "\", " << d.query << ", " << toInt(slot->normalization) << ")";
318 50 : }
319 :
320 50 : void PgQueryInterface::bindFullTextQuery(db::Binder &, StringStream &query, const db::Binder::FullTextQueryRef &d) {
321 50 : StringStream tmp;
322 50 : d.query.encode([&] (StringView str) {
323 75 : tmp << str;
324 75 : }, FullTextQuery::Postgresql);
325 50 : auto idx = push(tmp.str());
326 50 : query << " $" << idx << "::tsquery ";
327 50 : }
328 :
329 0 : void PgQueryInterface::bindIntVector(Binder &, StringStream &query, const Vector<int64_t> &vec) {
330 0 : query << "(";
331 0 : bool start = true;
332 0 : for (auto &it : vec) {
333 0 : if (start) { start = false; } else { query << ","; }
334 0 : query << it;
335 : }
336 0 : query << ")";
337 0 : }
338 :
339 0 : void PgQueryInterface::bindDoubleVector(Binder &b, StringStream &query, const Vector<double> &vec) {
340 0 : query << "(";
341 0 : bool start = true;
342 0 : for (auto &it : vec) {
343 0 : if (start) { start = false; } else { query << ","; }
344 0 : bindDouble(b, query, it);
345 : }
346 0 : query << ")";
347 0 : }
348 :
349 0 : void PgQueryInterface::bindStringVector(Binder &b, StringStream &query, const Vector<StringView> &vec) {
350 0 : query << "(";
351 0 : bool start = true;
352 0 : for (auto &it : vec) {
353 0 : if (start) { start = false; } else { query << ","; }
354 0 : bindStringView(b, query, it);
355 : }
356 0 : query << ")";
357 0 : }
358 :
359 1925 : void PgQueryInterface::clear() {
360 1925 : params.clear();
361 1925 : binary.clear();
362 1925 : }
363 :
364 375 : Handle::Handle(const Driver *d, Driver::Handle h) : SqlHandle(d), driver(d), handle(h) {
365 375 : if (h.get()) {
366 375 : auto c = d->getConnection(h);
367 375 : if (c.get()) {
368 375 : conn = c;
369 :
370 375 : performSimpleSelect("SELECT current_database();", [&, this] (db::Result &qResult) {
371 375 : if (!qResult.empty()) {
372 375 : dbName = qResult.current().toString(0).pdup();
373 : }
374 375 : });
375 : }
376 : }
377 375 : }
378 :
379 0 : Handle::operator bool() const {
380 0 : return conn.get() != nullptr;
381 : }
382 :
383 0 : Driver::Handle Handle::getHandle() const {
384 0 : return handle;
385 : }
386 :
387 0 : Driver::Connection Handle::getConnection() const {
388 0 : return conn;
389 : }
390 :
391 0 : void Handle::close() {
392 0 : conn = Driver::Connection(nullptr);
393 0 : }
394 :
395 3875 : void Handle::makeQuery(const stappler::Callback<void(sql::SqlQuery &)> &cb, const sql::QueryStorageHandle *s) {
396 3875 : PgQueryInterface interface(_driver, s);
397 3875 : db::sql::SqlQuery query(&interface, _driver);
398 3875 : cb(query);
399 3875 : }
400 :
401 3900 : bool Handle::selectQuery(const sql::SqlQuery &query, const stappler::Callback<bool(sql::Result &)> &cb,
402 : const Callback<void(const Value &)> &errCb) {
403 3900 : if (!conn.get() || getTransactionStatus() == db::TransactionStatus::Rollback) {
404 0 : return false;
405 : }
406 :
407 3900 : auto queryInterface = static_cast<PgQueryInterface *>(query.getInterface());
408 :
409 3900 : ExecParamData data(query);
410 3900 : ResultCursor res(driver, driver->exec(conn, query.getQuery().weak().data(), queryInterface->params.size(),
411 7800 : data.paramValues, data.paramLengths, data.paramFormats, 1));
412 3900 : if (!res.isSuccess()) {
413 0 : auto info = res.getInfo();
414 0 : info.setString(query.getQuery().str(), "query");
415 : #if DEBUG
416 0 : log::debug("pq::Handle", EncodeFormat::Pretty, info);
417 : #endif
418 0 : if (errCb) {
419 0 : errCb(info);
420 : }
421 0 : driver->getApplicationInterface()->debug("Database", "Fail to perform query", std::move(info));
422 0 : driver->getApplicationInterface()->error("Database", "Fail to perform query");
423 0 : cancelTransaction_pg();
424 0 : }
425 :
426 3900 : lastError = res.getError();
427 :
428 3900 : db::sql::Result ret(&res);
429 3900 : return cb(ret);
430 3900 : }
431 :
432 1300 : bool Handle::performSimpleQuery(const StringView &query, const Callback<void(const Value &)> &errCb) {
433 1300 : if (getTransactionStatus() == db::TransactionStatus::Rollback) {
434 0 : return false;
435 : }
436 :
437 1300 : ResultCursor res(driver, driver->exec(conn, query.data()));
438 1300 : lastError = res.getError();
439 1300 : if (!res.isSuccess()) {
440 0 : auto info = res.getInfo();
441 0 : info.setString(query, "query");
442 : #if DEBUG
443 0 : log::debug("pq::Handle", EncodeFormat::Pretty, info);
444 : #endif
445 0 : if (errCb) {
446 0 : errCb(info);
447 : }
448 0 : driver->getApplicationInterface()->debug("Database", "Fail to perform query", std::move(info));
449 0 : driver->getApplicationInterface()->error("Database", "Fail to perform query");
450 0 : cancelTransaction_pg();
451 0 : }
452 1300 : return res.isSuccess();
453 1300 : }
454 :
455 550 : bool Handle::performSimpleSelect(const StringView &query, const stappler::Callback<void(sql::Result &)> &cb,
456 : const Callback<void(const Value &)> &errCb) {
457 550 : if (getTransactionStatus() == db::TransactionStatus::Rollback) {
458 0 : return false;
459 : }
460 :
461 550 : ResultCursor res(driver, driver->exec(conn, query.data(), 0, nullptr, nullptr, nullptr, 1));
462 550 : lastError = res.getError();
463 :
464 550 : if (res.isSuccess()) {
465 550 : db::sql::Result ret(&res);
466 550 : cb(ret);
467 550 : return true;
468 550 : } else {
469 0 : auto info = res.getInfo();
470 0 : info.setString(query, "query");
471 : #if DEBUG
472 0 : log::debug("pq::Handle", EncodeFormat::Pretty, info);
473 : #endif
474 0 : if (errCb) {
475 0 : errCb(info);
476 : }
477 0 : driver->getApplicationInterface()->debug("Database", "Fail to perform query", std::move(info));
478 0 : driver->getApplicationInterface()->error("Database", "Fail to perform query");
479 0 : cancelTransaction_pg();
480 0 : }
481 :
482 0 : return false;
483 550 : }
484 :
485 0 : bool Handle::isSuccess() const {
486 0 : return ResultCursor::pgsql_is_success(lastError);
487 : }
488 :
489 375 : bool Handle::beginTransaction_pg(TransactionLevel l) {
490 375 : int64_t userId = _driver->getApplicationInterface()->getUserIdFromContext();
491 375 : int64_t now = stappler::Time::now().toMicros();
492 :
493 375 : if (transactionStatus != db::TransactionStatus::None) {
494 0 : log::error("pq::Handle", "Transaction already started");
495 0 : return false;
496 : }
497 :
498 375 : auto setVariables = [&, this] {
499 375 : performSimpleQuery(toString("SET LOCAL serenity.\"user\" = ", userId, ";SET LOCAL serenity.\"now\" = ", now, ";"));
500 375 : };
501 :
502 375 : switch (l) {
503 375 : case TransactionLevel::ReadCommited:
504 375 : if (performSimpleQuery("BEGIN ISOLATION LEVEL READ COMMITTED"_weak)) {
505 375 : setVariables();
506 375 : level = TransactionLevel::ReadCommited;
507 375 : transactionStatus = db::TransactionStatus::Commit;
508 375 : return true;
509 : }
510 0 : break;
511 0 : case TransactionLevel::RepeatableRead:
512 0 : if (performSimpleQuery("BEGIN ISOLATION LEVEL REPEATABLE READ"_weak)) {
513 0 : setVariables();
514 0 : level = TransactionLevel::RepeatableRead;
515 0 : transactionStatus = db::TransactionStatus::Commit;
516 0 : return true;
517 : }
518 0 : break;
519 0 : case TransactionLevel::Serialized:
520 0 : if (performSimpleQuery("BEGIN ISOLATION LEVEL SERIALIZABLE"_weak)) {
521 0 : setVariables();
522 0 : level = TransactionLevel::Serialized;
523 0 : transactionStatus = db::TransactionStatus::Commit;
524 0 : return true;
525 : }
526 0 : break;
527 0 : default:
528 0 : break;
529 : }
530 0 : return false;
531 : }
532 :
533 0 : void Handle::cancelTransaction_pg() {
534 0 : transactionStatus = db::TransactionStatus::Rollback;
535 0 : }
536 :
537 375 : bool Handle::endTransaction_pg() {
538 375 : switch (transactionStatus) {
539 375 : case db::TransactionStatus::Commit:
540 375 : transactionStatus = db::TransactionStatus::None;
541 375 : if (performSimpleQuery("COMMIT"_weak)) {
542 375 : finalizeBroadcast();
543 375 : return true;
544 : }
545 0 : break;
546 0 : case db::TransactionStatus::Rollback:
547 0 : transactionStatus = db::TransactionStatus::None;
548 0 : if (performSimpleQuery("ROLLBACK"_weak)) {
549 0 : finalizeBroadcast();
550 0 : return false;
551 : }
552 0 : break;
553 0 : default:
554 0 : break;
555 : }
556 0 : return false;
557 : }
558 :
559 : }
|
