001/* 002 * Copyright 2015-2022 Transmogrify LLC, 2022-2026 Revetware LLC. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016 017package com.pyranid; 018 019import org.jspecify.annotations.NonNull; 020import org.jspecify.annotations.Nullable; 021 022import javax.annotation.concurrent.NotThreadSafe; 023import javax.annotation.concurrent.ThreadSafe; 024import javax.sql.DataSource; 025import java.sql.Connection; 026import java.sql.DatabaseMetaData; 027import java.sql.PreparedStatement; 028import java.sql.ResultSet; 029import java.sql.SQLException; 030import java.sql.SQLFeatureNotSupportedException; 031import java.sql.Types; 032import java.time.Duration; 033import java.time.ZoneId; 034import java.util.ArrayDeque; 035import java.util.ArrayList; 036import java.util.Arrays; 037import java.util.Collection; 038import java.util.Deque; 039import java.util.HashSet; 040import java.util.LinkedHashMap; 041import java.util.LinkedHashSet; 042import java.util.List; 043import java.util.Locale; 044import java.util.Map; 045import java.util.Optional; 046import java.util.Queue; 047import java.util.Set; 048import java.util.Spliterator; 049import java.util.Spliterators; 050import java.util.concurrent.atomic.AtomicReference; 051import java.util.concurrent.atomic.AtomicLong; 052import java.util.concurrent.locks.ReentrantLock; 053import java.util.function.Consumer; 054import java.util.function.Function; 055import java.util.function.UnaryOperator; 056import java.util.UUID; 057import java.util.logging.Level; 058import java.util.logging.Logger; 059import java.util.regex.Pattern; 060import java.util.stream.Collectors; 061import java.util.stream.Stream; 062import java.util.stream.StreamSupport; 063 064import static java.lang.String.format; 065import static java.lang.System.nanoTime; 066import static java.util.Objects.requireNonNull; 067 068/** 069 * Main class for performing database access operations. 070 * 071 * @author <a href="https://www.revetkn.com">Mark Allen</a> 072 * @since 1.0.0 073 */ 074@ThreadSafe 075public final class Database { 076 @NonNull 077 private static final ThreadLocal<Deque<Transaction>> TRANSACTION_STACK_HOLDER; 078 private static final int DEFAULT_PARSED_SQL_CACHE_CAPACITY = 1024; 079 private static final int MAX_DIAGNOSTIC_MESSAGE_LENGTH = 1024; 080 private static final int MAX_DIAGNOSTIC_PARAMETER_LENGTH = 256; 081 private static final int MAX_DIAGNOSTIC_PARAMETERS_LENGTH = 2048; 082 private static final int MAX_DIAGNOSTIC_SQL_LENGTH = 2048; 083 @NonNull 084 private static final String TRUNCATED_SUFFIX = "... (truncated)"; 085 @NonNull 086 private static final Pattern DIAGNOSTIC_WHITESPACE_PATTERN = Pattern.compile("\\s+"); 087 088 static { 089 TRANSACTION_STACK_HOLDER = new ThreadLocal<>(); 090 } 091 092 @NonNull 093 private final DataSource dataSource; 094 @NonNull 095 private final AtomicReference<@Nullable DatabaseType> databaseType; 096 @NonNull 097 private final AtomicReference<@Nullable DatabaseDialect> databaseDialect; 098 @NonNull 099 private final ThreadLocal<Connection> databaseTypeDetectionConnectionHolder; 100 @NonNull 101 private final ZoneId timeZone; 102 @NonNull 103 private final AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy; 104 @NonNull 105 private final InstanceProvider instanceProvider; 106 @NonNull 107 private final PreparedStatementBinder preparedStatementBinder; 108 @NonNull 109 private final ResultSetMapper resultSetMapper; 110 @NonNull 111 private final StatementLogger statementLogger; 112 @NonNull 113 private final ParameterRedactor parameterRedactor; 114 @NonNull 115 private final MetricsCollectorDispatcher metricsCollectorDispatcher; 116 @Nullable 117 private final Duration queryTimeout; 118 @Nullable 119 private final Integer fetchSize; 120 @Nullable 121 private final Integer maxRows; 122 @Nullable 123 private final Map<String, ParsedSqlVariants> parsedSqlCache; 124 @NonNull 125 private final AtomicLong defaultIdGenerator; 126 @NonNull 127 private final Logger logger; 128 129 @NonNull 130 private volatile DatabaseOperationSupportStatus executeLargeBatchSupported; 131 @NonNull 132 private volatile DatabaseOperationSupportStatus executeLargeUpdateSupported; 133 134 private Database(@NonNull Builder builder) { 135 requireNonNull(builder); 136 137 this.dataSource = requireNonNull(builder.dataSource); 138 this.databaseType = new AtomicReference<>(builder.databaseType); 139 this.databaseDialect = new AtomicReference<>(builder.databaseType == null ? null : builder.databaseType.dialect()); 140 this.databaseTypeDetectionConnectionHolder = new ThreadLocal<>(); 141 this.timeZone = builder.timeZone == null ? ZoneId.systemDefault() : builder.timeZone; 142 this.ambiguousTimestampBindingStrategy = builder.ambiguousTimestampBindingStrategy == null 143 ? AmbiguousTimestampBindingStrategy.TIMESTAMP_WITH_TIME_ZONE 144 : builder.ambiguousTimestampBindingStrategy; 145 this.instanceProvider = builder.instanceProvider == null ? new InstanceProvider() {} : builder.instanceProvider; 146 this.preparedStatementBinder = builder.preparedStatementBinder == null ? PreparedStatementBinder.withDefaultConfiguration() : builder.preparedStatementBinder; 147 this.resultSetMapper = builder.resultSetMapper == null ? ResultSetMapper.withDefaultConfiguration() : builder.resultSetMapper; 148 this.statementLogger = builder.statementLogger == null ? (statementLog) -> {} : builder.statementLogger; 149 this.parameterRedactor = builder.parameterRedactor == null ? ParameterRedactor.none() : builder.parameterRedactor; 150 this.metricsCollectorDispatcher = new MetricsCollectorDispatcher(builder.metricsCollector); 151 this.queryTimeout = validateQueryTimeout(builder.queryTimeout); 152 this.fetchSize = validateNonNegativeStatementSetting("fetchSize", builder.fetchSize); 153 this.maxRows = validateNonNegativeStatementSetting("maxRows", builder.maxRows); 154 if (builder.parsedSqlCacheCapacity != null && builder.parsedSqlCacheCapacity < 0) 155 throw new IllegalArgumentException("parsedSqlCacheCapacity must be >= 0"); 156 157 int parsedSqlCacheCapacity = builder.parsedSqlCacheCapacity == null 158 ? DEFAULT_PARSED_SQL_CACHE_CAPACITY 159 : builder.parsedSqlCacheCapacity; 160 this.parsedSqlCache = parsedSqlCacheCapacity == 0 ? null : new ConcurrentLruMap<>(parsedSqlCacheCapacity); 161 this.defaultIdGenerator = new AtomicLong(); 162 this.logger = Logger.getLogger(getClass().getName()); 163 this.executeLargeBatchSupported = DatabaseOperationSupportStatus.UNKNOWN; 164 this.executeLargeUpdateSupported = DatabaseOperationSupportStatus.UNKNOWN; 165 } 166 167 /** 168 * Provides a {@link Database} builder for the given {@link DataSource}. 169 * 170 * @param dataSource data source used to create the {@link Database} builder 171 * @return a {@link Database} builder 172 */ 173 @NonNull 174 public static Builder withDataSource(@NonNull DataSource dataSource) { 175 requireNonNull(dataSource); 176 return new Builder(dataSource); 177 } 178 179 /** 180 * Gets a reference to the current transaction, if any. 181 * 182 * @return the current transaction 183 */ 184 @NonNull 185 public Optional<Transaction> currentTransaction() { 186 @Nullable Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get(); 187 Transaction transaction = transactionStack == null || transactionStack.isEmpty() ? null : transactionStack.peek(); 188 189 if (transaction == null || !isTransactionOwnedByThisDatabase(transaction)) 190 return Optional.empty(); 191 192 return Optional.of(transaction); 193 } 194 195 @NonNull 196 private Optional<Transaction> currentTransactionForDatabaseOperation() { 197 @Nullable Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get(); 198 Transaction transaction = transactionStack == null || transactionStack.isEmpty() ? null : transactionStack.peek(); 199 200 if (transaction == null) 201 return Optional.empty(); 202 203 if (!isTransactionOwnedByThisDatabase(transaction)) 204 throw wrongDatabaseTransactionException(transaction); 205 206 return Optional.of(transaction); 207 } 208 209 static boolean hasAmbientTransaction() { 210 Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get(); 211 return transactionStack != null && !transactionStack.isEmpty(); 212 } 213 214 @NonNull 215 private Deque<Transaction> transactionStackForPush() { 216 Deque<Transaction> transactionStack = TRANSACTION_STACK_HOLDER.get(); 217 218 if (transactionStack == null) { 219 transactionStack = new ArrayDeque<>(); 220 TRANSACTION_STACK_HOLDER.set(transactionStack); 221 } 222 223 return transactionStack; 224 } 225 226 private boolean isTransactionOwnedByThisDatabase(@NonNull Transaction transaction) { 227 requireNonNull(transaction); 228 return transaction.isOwnedBy(getDataSource()); 229 } 230 231 @NonNull 232 private DatabaseException wrongDatabaseTransactionException(@NonNull Transaction transaction) { 233 requireNonNull(transaction); 234 return new DatabaseException(format("Transaction %s belongs to a different %s than this %s. " 235 + "Use the %s instance that created the transaction, or explicitly participate only with a transaction " 236 + "created from the same %s.", 237 transaction.id(), DataSource.class.getSimpleName(), Database.class.getSimpleName(), 238 Database.class.getSimpleName(), DataSource.class.getSimpleName())); 239 } 240 241 /** 242 * Performs an operation transactionally. 243 * <p> 244 * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}. 245 * <p> 246 * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do 247 * not automatically join an outer transaction. Use {@link #participate(Transaction, TransactionalOperation)} to join an 248 * existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a 249 * {@link Database} using a different {@link DataSource} fails fast instead of joining it. 250 * 251 * @param transactionalOperation the operation to perform transactionally 252 */ 253 public void transaction(@NonNull TransactionalOperation transactionalOperation) { 254 requireNonNull(transactionalOperation); 255 256 transaction(() -> { 257 transactionalOperation.perform(); 258 return Optional.empty(); 259 }); 260 } 261 262 /** 263 * Performs an operation transactionally with the given options. 264 * <p> 265 * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}. 266 * <p> 267 * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do 268 * not automatically join an outer transaction. Use {@link #participate(Transaction, TransactionalOperation)} to join an 269 * existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a 270 * {@link Database} using a different {@link DataSource} fails fast instead of joining it. 271 * 272 * @param transactionOptions options to apply to the transaction 273 * @param transactionalOperation the operation to perform transactionally 274 * @since 4.2.0 275 */ 276 public void transaction(@NonNull TransactionOptions transactionOptions, 277 @NonNull TransactionalOperation transactionalOperation) { 278 requireNonNull(transactionOptions); 279 requireNonNull(transactionalOperation); 280 281 transaction(transactionOptions, () -> { 282 transactionalOperation.perform(); 283 return Optional.empty(); 284 }); 285 } 286 287 /** 288 * Performs an operation transactionally and optionally returns a value. 289 * <p> 290 * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}. 291 * <p> 292 * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do 293 * not automatically join an outer transaction. Use {@link #participate(Transaction, ReturningTransactionalOperation)} to 294 * join an existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a 295 * {@link Database} using a different {@link DataSource} fails fast instead of joining it. 296 * 297 * @param transactionalOperation the operation to perform transactionally 298 * @param <T> the type to be returned 299 * @return the result of the transactional operation 300 */ 301 @NonNull 302 public <T> Optional<T> transaction(@NonNull ReturningTransactionalOperation<T> transactionalOperation) { 303 requireNonNull(transactionalOperation); 304 return transaction(TransactionOptions.DEFAULT, transactionalOperation); 305 } 306 307 /** 308 * Performs an operation transactionally with the given options, optionally returning a value. 309 * <p> 310 * The transaction will be automatically rolled back if an exception bubbles out of {@code transactionalOperation}. 311 * <p> 312 * Nested calls to {@code transaction(...)} are independent transactions with independent JDBC connections; they do 313 * not automatically join an outer transaction. Use {@link #participate(Transaction, ReturningTransactionalOperation)} to 314 * join an existing transaction explicitly. A transaction is scoped to the {@link DataSource} instance that created it; a 315 * {@link Database} using a different {@link DataSource} fails fast instead of joining it. 316 * 317 * @param transactionOptions options to apply to the transaction 318 * @param transactionalOperation the operation to perform transactionally 319 * @param <T> the type to be returned 320 * @return the result of the transactional operation 321 * @since 4.2.0 322 */ 323 @NonNull 324 public <T> Optional<T> transaction(@NonNull TransactionOptions transactionOptions, 325 @NonNull ReturningTransactionalOperation<T> transactionalOperation) { 326 requireNonNull(transactionOptions); 327 requireNonNull(transactionalOperation); 328 329 Transaction transaction = new Transaction(dataSource, transactionOptions, getMetricsCollectorDispatcher(), 330 peekDatabaseType(), this::getDatabaseType); 331 Deque<Transaction> transactionStack = transactionStackForPush(); 332 transactionStack.push(transaction); 333 boolean committed = false; 334 boolean commitFailed = false; 335 boolean rollbackFailed = false; 336 boolean rollbackAttempted = false; 337 Throwable thrown = null; 338 long transactionStartTime = nanoTime(); 339 getMetricsCollectorDispatcher().didEnterTransactionClosure(transaction, transactionOptions.getIsolation(), transaction.getDatabaseType()); 340 341 try { 342 Optional<T> returnValue = transactionalOperation.perform(); 343 344 // Safeguard in case user code accidentally returns null instead of Optional.empty() 345 if (returnValue == null) 346 returnValue = Optional.empty(); 347 348 transaction.getConnectionLock().lock(); 349 350 try { 351 // A failed physical begin is terminal for this transaction even if application code caught its first 352 // manifestation. Never commit, or report success from a rollback-only path, through a partially initialized 353 // connection. 354 transaction.throwPhysicalTransactionBeginFailureIfPresent(); 355 356 if (transaction.isRollbackOnly()) { 357 rollbackAttempted = true; 358 transaction.rollback(); 359 } else { 360 try { 361 transaction.commit(); 362 } catch (RuntimeException | Error e) { 363 commitFailed = true; 364 throw e; 365 } 366 committed = true; 367 } 368 369 transaction.markCompleted(); 370 } finally { 371 transaction.getConnectionLock().unlock(); 372 } 373 374 return returnValue; 375 } catch (RuntimeException e) { 376 thrown = e; 377 if (rollbackAttempted) { 378 rollbackFailed = true; 379 markTransactionCompleted(transaction); 380 } else { 381 rollbackFailed = rollbackTransactionAfterFailure(transaction, e); 382 } 383 384 restoreInterruptIfNeeded(e); 385 throw e; 386 } catch (Error e) { 387 thrown = e; 388 if (rollbackAttempted) { 389 rollbackFailed = true; 390 markTransactionCompleted(transaction); 391 } else { 392 rollbackFailed = rollbackTransactionAfterFailure(transaction, e); 393 } 394 395 restoreInterruptIfNeeded(e); 396 throw e; 397 } catch (Throwable t) { 398 RuntimeException wrapped = new RuntimeException(t); 399 thrown = wrapped; 400 if (rollbackAttempted) { 401 rollbackFailed = true; 402 markTransactionCompleted(transaction); 403 } else { 404 rollbackFailed = rollbackTransactionAfterFailure(transaction, wrapped); 405 } 406 407 restoreInterruptIfNeeded(t); 408 throw wrapped; 409 } finally { 410 transactionStack.pop(); 411 412 // Ensure txn stack is fully cleaned up 413 if (transactionStack.isEmpty()) 414 TRANSACTION_STACK_HOLDER.remove(); 415 416 Throwable cleanupFailure = null; 417 boolean hadPhysicalTransaction = false; 418 boolean physicalTransactionBeganSuccessfully = false; 419 boolean completionFailed = commitFailed || rollbackFailed; 420 boolean commitSerializationFailureRolledBack = commitFailed && !rollbackFailed 421 && transaction.didCommitFailWithSerializationFailure(); 422 boolean transactionOutcomeIndeterminate = 423 (commitFailed && !commitSerializationFailureRolledBack) 424 || (rollbackFailed && transaction.couldApplicationWorkHaveExecuted()); 425 boolean discardConnection = completionFailed || transaction.didPhysicalTransactionBeginFail(); 426 427 try { 428 transaction.getConnectionLock().lock(); 429 430 try { 431 hadPhysicalTransaction = transaction.hasConnection(); 432 physicalTransactionBeganSuccessfully = transaction.didPhysicalTransactionBeginSuccessfully(); 433 434 if (!transaction.isCompleted()) 435 transaction.markCompleted(); 436 437 cleanupFailure = cleanupCompletedTransactionConnection(transaction, cleanupFailure, discardConnection); 438 } finally { 439 transaction.getConnectionLock().unlock(); 440 } 441 } finally { 442 if (transactionOutcomeIndeterminate && thrown instanceof DatabaseException) 443 ((DatabaseException) thrown).markTransactionOutcomeIndeterminate(); 444 445 // Execute any user-supplied post-execution hooks 446 for (Consumer<TransactionResult> postTransactionOperation : transaction.getPostTransactionOperations()) { 447 long postTransactionStartTime = nanoTime(); 448 Throwable postTransactionThrowable = null; 449 TransactionResult transactionResult = transactionResult(committed, transactionOutcomeIndeterminate); 450 try { 451 postTransactionOperation.accept(transactionResult); 452 } catch (Throwable cleanupException) { 453 postTransactionThrowable = cleanupException; 454 PostTransactionOperationException postTransactionOperationException = 455 new PostTransactionOperationException(transactionResult, cleanupException); 456 457 if (cleanupFailure == null) 458 cleanupFailure = postTransactionOperationException; 459 else 460 cleanupFailure.addSuppressed(postTransactionOperationException); 461 } finally { 462 getMetricsCollectorDispatcher().didRunPostTransactionOperation(transaction, transactionResult, transaction.getDatabaseType(), 463 Duration.ofNanos(nanoTime() - postTransactionStartTime), postTransactionThrowable); 464 } 465 } 466 } 467 468 Throwable exitThrown = thrown == null ? cleanupFailure : thrown; 469 getMetricsCollectorDispatcher().didExitTransactionClosure(transaction, 470 transactionClosureOutcome(committed, hadPhysicalTransaction, physicalTransactionBeganSuccessfully, 471 transactionOutcomeIndeterminate), 472 transaction.getDatabaseType(), Duration.ofNanos(nanoTime() - transactionStartTime), exitThrown); 473 474 if (cleanupFailure != null) { 475 if (thrown != null) { 476 if (thrown != cleanupFailure) 477 thrown.addSuppressed(cleanupFailure); 478 } else if (cleanupFailure instanceof RuntimeException) { 479 if (committed && cleanupFailure instanceof DatabaseException) 480 ((DatabaseException) cleanupFailure).markTransactionOutcomeCommitted(); 481 throw (RuntimeException) cleanupFailure; 482 } else if (cleanupFailure instanceof Error) { 483 throw (Error) cleanupFailure; 484 } else { 485 throw new RuntimeException(cleanupFailure); 486 } 487 } 488 } 489 } 490 491 /** 492 * Performs an operation transactionally, retrying according to the given retry policy. 493 * <p> 494 * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure 495 * unless they are safe to repeat. 496 * <p> 497 * Pyranid consults the retry policy only when replay is known to be safe. This includes a failure before a physical 498 * transaction becomes available to application work, even if best-effort rollback of that begin candidate fails, because 499 * the candidate is discarded. It also includes a recognized serialization failure reported by physical commit when the 500 * follow-up rollback succeeds. Other commit failures and rollback failures after application work could execute are 501 * terminal because their outcome is indeterminate. 502 * <p> 503 * Unlike {@link #transaction(TransactionalOperation)} and related transaction methods, retrying methods return 504 * {@link TransactionRetryResult} so callers can inspect failures that were recovered before success. 505 * <p> 506 * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot 507 * restart the outer transaction safely. 508 * 509 * @param retryPolicy retry policy to apply 510 * @param transactionalOperation the operation to perform transactionally 511 * @return retry result containing any failures retried before success 512 * @since 4.4.0 513 */ 514 @NonNull 515 public TransactionRetryResult<Void> transactionWithRetry(@NonNull RetryPolicy retryPolicy, 516 @NonNull TransactionalOperation transactionalOperation) { 517 requireNonNull(retryPolicy); 518 requireNonNull(transactionalOperation); 519 520 return transactionWithRetry(retryPolicy, () -> { 521 transactionalOperation.perform(); 522 return Optional.<Void>empty(); 523 }); 524 } 525 526 /** 527 * Performs an operation transactionally with the given options, retrying according to the given retry policy. 528 * <p> 529 * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure 530 * unless they are safe to repeat. 531 * <p> 532 * Pyranid consults the retry policy only when replay is known to be safe. This includes a failure before a physical 533 * transaction becomes available to application work, even if best-effort rollback of that begin candidate fails, because 534 * the candidate is discarded. It also includes a recognized serialization failure reported by physical commit when the 535 * follow-up rollback succeeds. Other commit failures and rollback failures after application work could execute are 536 * terminal because their outcome is indeterminate. 537 * <p> 538 * Unlike {@link #transaction(TransactionOptions, TransactionalOperation)} and related transaction methods, retrying 539 * methods return {@link TransactionRetryResult} so callers can inspect failures that were recovered before success. 540 * <p> 541 * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot 542 * restart the outer transaction safely. 543 * 544 * @param retryPolicy retry policy to apply 545 * @param transactionOptions options to apply to each transaction attempt 546 * @param transactionalOperation the operation to perform transactionally 547 * @return retry result containing any failures retried before success 548 * @since 4.4.0 549 */ 550 @NonNull 551 public TransactionRetryResult<Void> transactionWithRetry(@NonNull RetryPolicy retryPolicy, 552 @NonNull TransactionOptions transactionOptions, 553 @NonNull TransactionalOperation transactionalOperation) { 554 requireNonNull(retryPolicy); 555 requireNonNull(transactionOptions); 556 requireNonNull(transactionalOperation); 557 558 return transactionWithRetry(retryPolicy, transactionOptions, () -> { 559 transactionalOperation.perform(); 560 return Optional.<Void>empty(); 561 }); 562 } 563 564 /** 565 * Performs an operation transactionally and optionally returns a value, retrying according to the given retry policy. 566 * <p> 567 * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure 568 * unless they are safe to repeat. 569 * <p> 570 * Pyranid consults the retry policy only when replay is known to be safe. This includes a failure before a physical 571 * transaction becomes available to application work, even if best-effort rollback of that begin candidate fails, because 572 * the candidate is discarded. It also includes a recognized serialization failure reported by physical commit when the 573 * follow-up rollback succeeds. Other commit failures and rollback failures after application work could execute are 574 * terminal because their outcome is indeterminate. 575 * <p> 576 * Unlike {@link #transaction(ReturningTransactionalOperation)} and related transaction methods, retrying methods return 577 * {@link TransactionRetryResult} so callers can inspect failures that were recovered before success. 578 * <p> 579 * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot 580 * restart the outer transaction safely. 581 * 582 * @param retryPolicy retry policy to apply 583 * @param transactionalOperation the operation to perform transactionally 584 * @param <T> the type to be returned 585 * @return retry result containing the successful transaction value and any failures retried before success 586 * @since 4.4.0 587 */ 588 @NonNull 589 public <T> TransactionRetryResult<T> transactionWithRetry(@NonNull RetryPolicy retryPolicy, 590 @NonNull ReturningTransactionalOperation<T> transactionalOperation) { 591 requireNonNull(retryPolicy); 592 requireNonNull(transactionalOperation); 593 594 return transactionWithRetry(retryPolicy, TransactionOptions.DEFAULT, transactionalOperation); 595 } 596 597 /** 598 * Performs an operation transactionally with the given options and optionally returns a value, retrying according to the 599 * given retry policy. 600 * <p> 601 * The entire transaction closure may run more than once. Keep non-idempotent external side effects outside the closure 602 * unless they are safe to repeat. 603 * <p> 604 * Pyranid consults the retry policy only when replay is known to be safe. This includes a failure before a physical 605 * transaction becomes available to application work, even if best-effort rollback of that begin candidate fails, because 606 * the candidate is discarded. It also includes a recognized serialization failure reported by physical commit when the 607 * follow-up rollback succeeds. Other commit failures and rollback failures after application work could execute are 608 * terminal because their outcome is indeterminate. 609 * <p> 610 * Unlike {@link #transaction(TransactionOptions, ReturningTransactionalOperation)} and related transaction methods, 611 * retrying methods return {@link TransactionRetryResult} so callers can inspect failures that were recovered before 612 * success. 613 * <p> 614 * This method fails fast if called inside an active transaction for this {@code Database}. Retrying a nested unit cannot 615 * restart the outer transaction safely. 616 * 617 * @param retryPolicy retry policy to apply 618 * @param transactionOptions options to apply to each transaction attempt 619 * @param transactionalOperation the operation to perform transactionally 620 * @param <T> the type to be returned 621 * @return retry result containing the successful transaction value and any failures retried before success 622 * @since 4.4.0 623 */ 624 @NonNull 625 public <T> TransactionRetryResult<T> transactionWithRetry(@NonNull RetryPolicy retryPolicy, 626 @NonNull TransactionOptions transactionOptions, 627 @NonNull ReturningTransactionalOperation<T> transactionalOperation) { 628 requireNonNull(retryPolicy); 629 requireNonNull(transactionOptions); 630 requireNonNull(transactionalOperation); 631 632 if (currentTransaction().isPresent()) 633 throw new IllegalStateException("transactionWithRetry must not be called within an existing transaction"); 634 635 List<DatabaseException> priorFailures = new ArrayList<>(); 636 637 for (int attempt = 1; attempt <= retryPolicy.getMaxAttempts(); ++attempt) { 638 try { 639 return TransactionRetryResult.of(transaction(transactionOptions, transactionalOperation), priorFailures); 640 } catch (DatabaseException e) { 641 if (e.isTransactionOutcomeRetryUnsafe()) { 642 suppressPriorFailures(e, priorFailures); 643 throw e; 644 } 645 646 boolean finalAttempt = attempt == retryPolicy.getMaxAttempts(); 647 648 if (finalAttempt) { 649 suppressPriorFailures(e, priorFailures); 650 throw e; 651 } 652 653 Boolean retryable; 654 655 try { 656 retryable = retryPolicy.getCondition().shouldRetry(e); 657 } catch (RuntimeException | Error conditionFailure) { 658 suppressRetryFailures(conditionFailure, e, priorFailures); 659 throw conditionFailure; 660 } 661 662 if (retryable == null) { 663 NullPointerException nullConditionFailure = 664 new NullPointerException("RetryPolicy.Condition returned null"); 665 suppressRetryFailures(nullConditionFailure, e, priorFailures); 666 throw nullConditionFailure; 667 } 668 669 if (!retryable) { 670 suppressPriorFailures(e, priorFailures); 671 throw e; 672 } 673 674 priorFailures.add(e); 675 676 Duration delay; 677 678 try { 679 delay = retryPolicy.getBackoff().delayAfterFailedAttempt(attempt, e); 680 } catch (RuntimeException | Error backoffFailure) { 681 suppressPriorFailures(backoffFailure, priorFailures); 682 throw backoffFailure; 683 } 684 685 if (delay == null) { 686 NullPointerException nullBackoffFailure = new NullPointerException("RetryPolicy.Backoff returned null"); 687 suppressPriorFailures(nullBackoffFailure, priorFailures); 688 throw nullBackoffFailure; 689 } 690 691 if (delay.isNegative()) { 692 IllegalArgumentException negativeBackoffFailure = 693 new IllegalArgumentException("RetryPolicy.Backoff returned a negative delay"); 694 suppressPriorFailures(negativeBackoffFailure, priorFailures); 695 throw negativeBackoffFailure; 696 } 697 698 try { 699 sleepBackoff(delay); 700 } catch (InterruptedException interruptedException) { 701 Thread.currentThread().interrupt(); 702 suppressPriorFailures(e, priorFailures); 703 e.addSuppressed(interruptedException); 704 throw e; 705 } 706 } 707 } 708 709 throw new AssertionError("unreachable"); 710 } 711 712 private void sleepBackoff(@NonNull Duration delay) throws InterruptedException { 713 requireNonNull(delay); 714 715 if (delay.isZero()) 716 return; 717 718 long millis; 719 int nanos; 720 721 try { 722 millis = delay.toMillis(); 723 Duration remainder = delay.minusMillis(millis); 724 nanos = (int) Math.min(999_999L, Math.max(0L, remainder.toNanos())); 725 } catch (ArithmeticException e) { 726 millis = Long.MAX_VALUE; 727 nanos = 999_999; 728 } 729 730 Thread.sleep(millis, nanos); 731 } 732 733 private void suppressRetryFailures(@NonNull Throwable failure, 734 @NonNull DatabaseException currentFailure, 735 @NonNull List<@NonNull DatabaseException> priorFailures) { 736 requireNonNull(failure); 737 requireNonNull(currentFailure); 738 requireNonNull(priorFailures); 739 740 suppressPriorFailures(failure, priorFailures); 741 suppressIfDifferent(failure, currentFailure); 742 } 743 744 private void suppressPriorFailures(@NonNull Throwable failure, 745 @NonNull List<@NonNull DatabaseException> priorFailures) { 746 requireNonNull(failure); 747 requireNonNull(priorFailures); 748 749 for (DatabaseException priorFailure : priorFailures) 750 suppressIfDifferent(failure, priorFailure); 751 } 752 753 private void suppressIfDifferent(@NonNull Throwable failure, 754 @NonNull Throwable suppressed) { 755 requireNonNull(failure); 756 requireNonNull(suppressed); 757 758 if (failure != suppressed) 759 failure.addSuppressed(suppressed); 760 } 761 762 private boolean rollbackTransactionAfterFailure(@NonNull Transaction transaction, 763 @NonNull Throwable primary) { 764 requireNonNull(transaction); 765 requireNonNull(primary); 766 767 boolean rollbackFailed = false; 768 transaction.getConnectionLock().lock(); 769 770 try { 771 try { 772 if (transaction.isPhysicalRollbackPermitted()) { 773 try { 774 transaction.rollback(); 775 } catch (Throwable rollbackException) { 776 rollbackFailed = true; 777 if (primary != rollbackException) 778 primary.addSuppressed(rollbackException); 779 } 780 } 781 } finally { 782 transaction.markCompleted(); 783 } 784 } finally { 785 transaction.getConnectionLock().unlock(); 786 } 787 788 return rollbackFailed; 789 } 790 791 private void markTransactionCompleted(@NonNull Transaction transaction) { 792 requireNonNull(transaction); 793 transaction.getConnectionLock().lock(); 794 795 try { 796 transaction.markCompleted(); 797 } finally { 798 transaction.getConnectionLock().unlock(); 799 } 800 } 801 802 @Nullable 803 private Throwable cleanupCompletedTransactionConnection(@NonNull Transaction transaction, 804 @Nullable Throwable cleanupFailure, 805 boolean discardConnection) { 806 requireNonNull(transaction); 807 boolean abortRequired = discardConnection; 808 809 // A begin, commit, or rollback failure makes the connection unsafe to restore and reuse, even when the logical outcome 810 // is known. When the outcome is unknown, restoring auto-commit to true is especially dangerous because it can commit 811 // work that Pyranid meant to roll back. 812 if (!abortRequired) { 813 try { 814 transaction.restoreTransactionIsolationIfNeeded(); 815 } catch (Throwable cleanupException) { 816 cleanupFailure = appendSuppressed(cleanupFailure, cleanupException); 817 abortRequired = true; 818 } 819 820 if (!abortRequired) { 821 try { 822 transaction.restoreReadOnlyIfNeeded(); 823 } catch (Throwable cleanupException) { 824 cleanupFailure = appendSuppressed(cleanupFailure, cleanupException); 825 abortRequired = true; 826 } 827 } 828 829 if (!abortRequired && transaction.getInitialAutoCommit().isPresent() && transaction.getInitialAutoCommit().get()) { 830 try { 831 // Autocommit was true initially, so restoring to true now that transaction has completed 832 transaction.setAutoCommit(true); 833 } catch (Throwable cleanupException) { 834 cleanupFailure = appendSuppressed(cleanupFailure, cleanupException); 835 abortRequired = true; 836 } 837 } 838 } 839 840 Connection connection = transaction.getExistingConnection().orElse(null); 841 842 if (connection != null) { 843 Duration heldDuration = transaction.getConnectionAcquiredAtNanos() 844 .map(acquiredAtNanos -> Duration.ofNanos(nanoTime() - acquiredAtNanos)) 845 .orElse(Duration.ZERO); 846 847 if (abortRequired) { 848 try { 849 abortConnection(connection); 850 } catch (Throwable cleanupException) { 851 cleanupFailure = appendSuppressed(cleanupFailure, cleanupException); 852 } 853 } 854 855 try { 856 closeConnection(connection); 857 getMetricsCollectorDispatcher().didReleaseTransactionConnection(transaction, transaction.getDatabaseType(), heldDuration); 858 } catch (Throwable cleanupException) { 859 getMetricsCollectorDispatcher().didFailToReleaseTransactionConnection(transaction, transaction.getDatabaseType(), heldDuration, cleanupException); 860 cleanupFailure = appendSuppressed(cleanupFailure, cleanupException); 861 } finally { 862 // A completed Transaction must never retain a discarded or closed connection handle, including when close fails. 863 transaction.clearConnection(); 864 } 865 } 866 867 return cleanupFailure; 868 } 869 870 @NonNull 871 private static Throwable appendSuppressed(@Nullable Throwable existing, 872 @NonNull Throwable additional) { 873 requireNonNull(additional); 874 875 if (existing == null) 876 return additional; 877 878 if (existing != additional) 879 existing.addSuppressed(additional); 880 881 return existing; 882 } 883 884 private void closeConnection(@NonNull Connection connection) { 885 requireNonNull(connection); 886 887 try { 888 connection.close(); 889 } catch (SQLException e) { 890 throw new DatabaseException("Unable to close database connection", e); 891 } 892 } 893 894 private void abortConnection(@NonNull Connection connection) { 895 requireNonNull(connection); 896 897 try { 898 connection.abort(Runnable::run); 899 } catch (SQLException e) { 900 throw new DatabaseException("Unable to abort database connection", e); 901 } 902 } 903 904 @NonNull 905 private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext, 906 @NonNull Throwable cause) { 907 requireNonNull(statementContext); 908 requireNonNull(cause); 909 910 if (cause instanceof DatabaseException databaseException && databaseException.areStatementDiagnosticsApplied()) 911 return databaseException; 912 913 String message = cause.getMessage(); 914 915 if (message == null || message.trim().length() == 0) 916 message = "Database operation failed"; 917 918 if (cause instanceof DatabaseException databaseException) { 919 SecureParameterSupport.DiagnosticScrub diagnosticScrub = diagnosticScrubForStatementContext(statementContext); 920 921 if (diagnosticScrub.redactor() == null && diagnosticScrub.needleRenderingFailureDescriptions().isEmpty()) 922 return databaseException; 923 924 return databaseExceptionWithStatementContext(statementContext, message, cause, diagnosticScrub); 925 } 926 927 return databaseExceptionWithStatementContext(statementContext, message, cause); 928 } 929 930 @NonNull 931 private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext, 932 @NonNull String message, 933 @NonNull Throwable cause) { 934 requireNonNull(statementContext); 935 requireNonNull(message); 936 requireNonNull(cause); 937 938 SecureParameterSupport.DiagnosticScrub diagnosticScrub = diagnosticScrubForStatementContext(statementContext); 939 return databaseExceptionWithStatementContext(statementContext, message, cause, diagnosticScrub); 940 } 941 942 @NonNull 943 private static DatabaseException databaseExceptionWithStatementContext(@NonNull StatementContext<?> statementContext, 944 @NonNull String message, 945 @NonNull Throwable cause, 946 SecureParameterSupport.@NonNull DiagnosticScrub diagnosticScrub) { 947 requireNonNull(statementContext); 948 requireNonNull(message); 949 requireNonNull(cause); 950 requireNonNull(diagnosticScrub); 951 952 UnaryOperator<String> diagnosticRedactor = diagnosticScrub.redactor(); 953 954 // Scrub the RAW message before whitespace-collapsing/truncation: bounding first would break needles 955 // containing consecutive whitespace and can cut a secret mid-way, leaving an unmatchable prefix 956 String scrubbedMessage = diagnosticRedactor == null ? message : diagnosticRedactor.apply(message); 957 958 StatementDiagnostic statementDiagnostic = statementDiagnostic(statementContext); 959 String diagnosticMessage = format("%s [%s]", 960 boundedDiagnosticMessage(scrubbedMessage), statementDiagnostic.diagnostic()); 961 DatabaseException databaseException = cause instanceof DatabaseException original 962 ? new DatabaseException(diagnosticMessage, original, diagnosticRedactor) 963 : new DatabaseException(diagnosticMessage, cause, 964 databaseDialectForException(statementContext, cause), diagnosticRedactor); 965 databaseException.markStatementDiagnosticsApplied(); 966 967 if (statementDiagnostic.parameterRenderingFailure() != null) 968 databaseException.addSuppressed(statementDiagnostic.parameterRenderingFailure()); 969 970 for (String failureDescription : diagnosticScrub.needleRenderingFailureDescriptions()) 971 databaseException.addSuppressed(new RuntimeException(format( 972 "Unable to render a secure parameter for diagnostic scrubbing: %s", failureDescription))); 973 974 return databaseException; 975 } 976 977 private static SecureParameterSupport.@NonNull DiagnosticScrub diagnosticScrubForStatementContext(@NonNull StatementContext<?> statementContext) { 978 requireNonNull(statementContext); 979 980 try { 981 return SecureParameterSupport.diagnosticScrubForParameters(statementContext.getParameters()); 982 } catch (Throwable ignored) { 983 // Per-needle failures are handled inside diagnosticScrubForParameters; reaching here would be 984 // catastrophic (a plain getter throwing) - fall back to no scrubbing rather than losing the exception 985 return SecureParameterSupport.DiagnosticScrub.NONE; 986 } 987 } 988 989 @NonNull 990 private static DatabaseDialect databaseDialectForException(@NonNull StatementContext<?> statementContext, 991 @NonNull Throwable cause) { 992 requireNonNull(statementContext); 993 requireNonNull(cause); 994 995 try { 996 return statementContext.getDatabaseDialect(); 997 } catch (Throwable ignored) { 998 return DatabaseDialect.forExceptionCause(cause); 999 } 1000 } 1001 1002 @NonNull 1003 private DatabaseException databaseExceptionWithRawConnectionContext(@Nullable Connection connection, 1004 @NonNull Exception cause) { 1005 requireNonNull(cause); 1006 1007 return new DatabaseException(cause.getMessage(), cause, databaseDialectForRawConnectionException(connection, cause)); 1008 } 1009 1010 @NonNull 1011 private DatabaseDialect databaseDialectForRawConnectionException(@Nullable Connection connection, 1012 @NonNull Throwable cause) { 1013 requireNonNull(cause); 1014 1015 if (connection != null) { 1016 try { 1017 return getDatabaseDialect(connection); 1018 } catch (Throwable ignored) { 1019 // Fall through to cause-based classification below. 1020 } 1021 } 1022 1023 return DatabaseDialect.forExceptionCause(cause); 1024 } 1025 1026 @NonNull 1027 private static StatementDiagnostic statementDiagnostic(@NonNull StatementContext<?> statementContext) { 1028 requireNonNull(statementContext); 1029 1030 Statement statement = statementContext.getStatement(); 1031 String parameters; 1032 Throwable parameterRenderingFailure = null; 1033 1034 try { 1035 parameters = boundedDiagnosticParameters(statementContext.getRedactedParameters()); 1036 } catch (Throwable t) { 1037 parameters = "<unavailable>"; 1038 parameterRenderingFailure = t; 1039 } 1040 1041 String diagnostic = format("statementId=%s, sql=%s, parameters=%s", 1042 statement.getId(), boundedSql(statement.getSql()), parameters); 1043 1044 if (parameterRenderingFailure != null) 1045 diagnostic = format("%s, parameterRenderingFailure=%s", diagnostic, 1046 boundedDiagnosticParameter(parameterRenderingFailureDiagnostic(parameterRenderingFailure))); 1047 1048 return new StatementDiagnostic(diagnostic, parameterRenderingFailure); 1049 } 1050 1051 private record StatementDiagnostic(@NonNull String diagnostic, 1052 @Nullable Throwable parameterRenderingFailure) {} 1053 1054 @NonNull 1055 private static String parameterRenderingFailureDiagnostic(@NonNull Throwable parameterRenderingFailure) { 1056 requireNonNull(parameterRenderingFailure); 1057 1058 String message; 1059 1060 try { 1061 message = parameterRenderingFailure.getMessage(); 1062 } catch (Throwable ignored) { 1063 message = null; 1064 } 1065 1066 if (message == null || message.trim().length() == 0) 1067 return parameterRenderingFailure.getClass().getName(); 1068 1069 return format("%s: %s", parameterRenderingFailure.getClass().getName(), message); 1070 } 1071 1072 @NonNull 1073 private static TransactionResult transactionResult(@NonNull Boolean committed, 1074 @NonNull Boolean transactionOutcomeIndeterminate) { 1075 requireNonNull(committed); 1076 requireNonNull(transactionOutcomeIndeterminate); 1077 1078 if (committed) 1079 return TransactionResult.COMMITTED; 1080 1081 return transactionOutcomeIndeterminate ? TransactionResult.IN_DOUBT : TransactionResult.ROLLED_BACK; 1082 } 1083 1084 private static MetricsCollector.TransactionClosureOutcome transactionClosureOutcome(@NonNull Boolean committed, 1085 @NonNull Boolean hadPhysicalTransaction, 1086 @NonNull Boolean physicalTransactionBeganSuccessfully, 1087 @NonNull Boolean transactionOutcomeIndeterminate) { 1088 requireNonNull(committed); 1089 requireNonNull(hadPhysicalTransaction); 1090 requireNonNull(physicalTransactionBeganSuccessfully); 1091 requireNonNull(transactionOutcomeIndeterminate); 1092 1093 if (!hadPhysicalTransaction) 1094 return MetricsCollector.TransactionClosureOutcome.NO_PHYSICAL_TX; 1095 1096 if (!physicalTransactionBeganSuccessfully) 1097 return MetricsCollector.TransactionClosureOutcome.FAILED; 1098 1099 if (committed) 1100 return MetricsCollector.TransactionClosureOutcome.COMMITTED; 1101 1102 return transactionOutcomeIndeterminate 1103 ? MetricsCollector.TransactionClosureOutcome.FAILED 1104 : MetricsCollector.TransactionClosureOutcome.ROLLED_BACK; 1105 } 1106 1107 @Nullable 1108 static Long sumBatchUpdateCounts(@NonNull List<Long> updateCounts) { 1109 requireNonNull(updateCounts); 1110 1111 long total = 0L; 1112 1113 for (Long updateCount : updateCounts) { 1114 if (updateCount == null || updateCount < 0L) 1115 return null; 1116 1117 try { 1118 total = Math.addExact(total, updateCount); 1119 } catch (ArithmeticException e) { 1120 return null; 1121 } 1122 } 1123 1124 return total; 1125 } 1126 1127 @NonNull 1128 private static String boundedDiagnosticMessage(@NonNull String message) { 1129 requireNonNull(message); 1130 1131 String compactMessage = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(message).replaceAll(" ").trim(); 1132 1133 if (compactMessage.length() <= MAX_DIAGNOSTIC_MESSAGE_LENGTH) 1134 return compactMessage; 1135 1136 int prefixLength = Math.max(0, MAX_DIAGNOSTIC_MESSAGE_LENGTH - TRUNCATED_SUFFIX.length()); 1137 return compactMessage.substring(0, prefixLength) + TRUNCATED_SUFFIX; 1138 } 1139 1140 @NonNull 1141 private static String boundedSql(@NonNull String sql) { 1142 requireNonNull(sql); 1143 1144 String compactSql = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(sql).replaceAll(" ").trim(); 1145 1146 if (compactSql.length() <= MAX_DIAGNOSTIC_SQL_LENGTH) 1147 return compactSql; 1148 1149 int prefixLength = Math.max(0, MAX_DIAGNOSTIC_SQL_LENGTH - TRUNCATED_SUFFIX.length()); 1150 return compactSql.substring(0, prefixLength) + TRUNCATED_SUFFIX; 1151 } 1152 1153 @NonNull 1154 private static String boundedDiagnosticParameters(@NonNull List<@Nullable Object> parameters) { 1155 requireNonNull(parameters); 1156 1157 StringBuilder parametersBuilder = new StringBuilder("["); 1158 1159 for (int i = 0; i < parameters.size(); ++i) { 1160 String separator = i == 0 ? "" : ", "; 1161 String renderedParameter = boundedDiagnosticParameter(parameters.get(i)); 1162 int requiredLength = parametersBuilder.length() + separator.length() + renderedParameter.length() + 1; 1163 1164 if (requiredLength > MAX_DIAGNOSTIC_PARAMETERS_LENGTH) { 1165 int availableLength = MAX_DIAGNOSTIC_PARAMETERS_LENGTH 1166 - parametersBuilder.length() 1167 - separator.length() 1168 - TRUNCATED_SUFFIX.length() 1169 - 1; 1170 1171 parametersBuilder.append(separator); 1172 1173 if (availableLength > 0) 1174 parametersBuilder.append(renderedParameter, 0, Math.min(availableLength, renderedParameter.length())); 1175 1176 parametersBuilder.append(TRUNCATED_SUFFIX).append(']'); 1177 if (parametersBuilder.length() > MAX_DIAGNOSTIC_PARAMETERS_LENGTH) { 1178 parametersBuilder.setLength(MAX_DIAGNOSTIC_PARAMETERS_LENGTH - 1); 1179 parametersBuilder.append(']'); 1180 } 1181 return parametersBuilder.toString(); 1182 } 1183 1184 parametersBuilder.append(separator).append(renderedParameter); 1185 } 1186 1187 parametersBuilder.append(']'); 1188 return parametersBuilder.toString(); 1189 } 1190 1191 @NonNull 1192 private static String boundedDiagnosticParameter(@Nullable Object parameter) { 1193 String renderedParameter = String.valueOf(parameter); 1194 String compactParameter = DIAGNOSTIC_WHITESPACE_PATTERN.matcher(renderedParameter).replaceAll(" ").trim(); 1195 1196 if (compactParameter.length() <= MAX_DIAGNOSTIC_PARAMETER_LENGTH) 1197 return compactParameter; 1198 1199 int prefixLength = Math.max(0, MAX_DIAGNOSTIC_PARAMETER_LENGTH - TRUNCATED_SUFFIX.length()); 1200 return compactParameter.substring(0, prefixLength) + TRUNCATED_SUFFIX; 1201 } 1202 1203 /** 1204 * Performs an operation in the context of a pre-existing transaction. 1205 * <p> 1206 * No commit or rollback on the transaction will occur when {@code transactionalOperation} completes. 1207 * <p> 1208 * However, if an exception bubbles out of {@code transactionalOperation}, the transaction will be marked as rollback-only. 1209 * <p> 1210 * The transaction must have been created by this {@link Database}, or by another {@link Database} using the same 1211 * {@link DataSource} instance. 1212 * <p> 1213 * If this thread is interrupted while waiting for another participant to release the transaction connection, Pyranid 1214 * restores the interrupt flag and throws {@link DatabaseException}. 1215 * 1216 * @param transaction the transaction in which to participate 1217 * @param transactionalOperation the operation that should participate in the transaction 1218 */ 1219 public void participate(@NonNull Transaction transaction, 1220 @NonNull TransactionalOperation transactionalOperation) { 1221 requireNonNull(transaction); 1222 requireNonNull(transactionalOperation); 1223 1224 participate(transaction, () -> { 1225 transactionalOperation.perform(); 1226 return Optional.empty(); 1227 }); 1228 } 1229 1230 /** 1231 * Performs an operation in the context of a pre-existing transaction, optionally returning a value. 1232 * <p> 1233 * No commit or rollback on the transaction will occur when {@code transactionalOperation} completes. 1234 * <p> 1235 * However, if an exception bubbles out of {@code transactionalOperation}, the transaction will be marked as rollback-only. 1236 * <p> 1237 * The transaction must have been created by this {@link Database}, or by another {@link Database} using the same 1238 * {@link DataSource} instance. 1239 * <p> 1240 * If this thread is interrupted while waiting for another participant to release the transaction connection, Pyranid 1241 * restores the interrupt flag and throws {@link DatabaseException}. 1242 * 1243 * @param transaction the transaction in which to participate 1244 * @param transactionalOperation the operation that should participate in the transaction 1245 * @param <T> the type to be returned 1246 * @return the result of the transactional operation 1247 */ 1248 @NonNull 1249 public <T> Optional<T> participate(@NonNull Transaction transaction, 1250 @NonNull ReturningTransactionalOperation<T> transactionalOperation) { 1251 requireNonNull(transaction); 1252 requireNonNull(transactionalOperation); 1253 1254 if (!isTransactionOwnedByThisDatabase(transaction)) 1255 throw wrongDatabaseTransactionException(transaction); 1256 1257 if (transaction.isCompleted()) 1258 throw new IllegalStateException(format("Transaction %s has already completed and cannot participate", transaction.id())); 1259 1260 Deque<Transaction> transactionStack = transactionStackForPush(); 1261 transactionStack.push(transaction); 1262 1263 try { 1264 Optional<T> returnValue = transactionalOperation.perform(); 1265 return returnValue == null ? Optional.empty() : returnValue; 1266 } catch (RuntimeException e) { 1267 setRollbackOnlyAfterParticipationFailure(transaction, e); 1268 restoreInterruptIfNeeded(e); 1269 throw e; 1270 } catch (Error e) { 1271 setRollbackOnlyAfterParticipationFailure(transaction, e); 1272 restoreInterruptIfNeeded(e); 1273 throw e; 1274 } catch (Throwable t) { 1275 RuntimeException wrapped = new RuntimeException(t); 1276 setRollbackOnlyAfterParticipationFailure(transaction, wrapped); 1277 restoreInterruptIfNeeded(t); 1278 throw wrapped; 1279 } finally { 1280 try { 1281 transactionStack.pop(); 1282 } finally { 1283 if (transactionStack.isEmpty()) 1284 TRANSACTION_STACK_HOLDER.remove(); 1285 } 1286 } 1287 } 1288 1289 private void setRollbackOnlyAfterParticipationFailure(@NonNull Transaction transaction, 1290 @NonNull Throwable primary) { 1291 requireNonNull(transaction); 1292 requireNonNull(primary); 1293 1294 try { 1295 transaction.setRollbackOnly(true); 1296 } catch (IllegalStateException e) { 1297 primary.addSuppressed(e); 1298 } 1299 } 1300 1301 /** 1302 * Creates a fluent builder for executing SQL. 1303 * <p> 1304 * Named parameters use the {@code :paramName} syntax and are bound via {@link Query#bind(String, Object)}. 1305 * Positional parameters via {@code ?} are not supported. 1306 * Pyranid ignores parameter-looking text inside SQL string literals, quoted identifiers, comments, PostgreSQL 1307 * dollar-quoted strings, and SQL Server-style bracket-quoted identifiers. PostgreSQL JSONB/hstore {@code ?}, 1308 * {@code ?|}, and {@code ?&} operators are supported; when running against {@link DatabaseType#POSTGRESQL}, Pyranid 1309 * emits pgjdbc's escaped {@code ??} form automatically. Unterminated quotes and comments fail fast. 1310 * <p> 1311 * Example: 1312 * <pre>{@code 1313 * Optional<Employee> employee = database.query("SELECT * FROM employee WHERE id = :id") 1314 * .bind("id", 42) 1315 * .fetchObject(Employee.class); 1316 * }</pre> 1317 * 1318 * @param sql SQL containing {@code :paramName} placeholders 1319 * @return a fluent builder for binding parameters and executing 1320 * @since 4.0.0 1321 */ 1322 @NonNull 1323 public Query query(@NonNull String sql) { 1324 requireNonNull(sql); 1325 return new DefaultQuery(this, sql); 1326 } 1327 1328 /** 1329 * Performs a portable connectivity check using JDBC {@link Connection#isValid(int)}. 1330 * <p> 1331 * This method borrows a fresh connection from this database's {@link DataSource}, calls 1332 * {@link Connection#isValid(int)}, and closes the connection before returning. It does <strong>not</strong> 1333 * participate in an active Pyranid transaction, if one exists. 1334 * <p> 1335 * JDBC accepts timeout values in whole seconds. Positive sub-second durations are rounded up to one second; 1336 * {@link Duration#ZERO} passes a timeout of {@code 0} to the driver. 1337 * 1338 * @param timeout maximum time to wait for driver validation 1339 * @throws IllegalArgumentException if {@code timeout} is negative or too large for JDBC's integer-second timeout 1340 * @throws DatabaseException if connection acquisition fails, validation throws, or the driver reports the 1341 * connection is not valid 1342 * @since 4.2.0 1343 */ 1344 public void performHealthCheck(@NonNull Duration timeout) { 1345 requireNonNull(timeout); 1346 int timeoutSeconds = healthCheckTimeoutSeconds(timeout); 1347 1348 performRawConnectionOperation(connection -> { 1349 boolean valid; 1350 1351 try { 1352 valid = connection.isValid(timeoutSeconds); 1353 } catch (SQLException e) { 1354 throw new DatabaseException("Unable to perform database health check", e); 1355 } 1356 1357 if (!valid) 1358 throw new DatabaseException("Database health check failed: connection is not valid"); 1359 1360 return Optional.empty(); 1361 }, false); 1362 } 1363 1364 private static void restoreInterruptIfNeeded(@NonNull Throwable throwable) { 1365 requireNonNull(throwable); 1366 1367 Throwable current = throwable; 1368 1369 while (current != null) { 1370 if (current instanceof InterruptedException) { 1371 Thread.currentThread().interrupt(); 1372 return; 1373 } 1374 1375 current = current.getCause(); 1376 } 1377 } 1378 1379 private static void lockInterruptibly(@NonNull ReentrantLock lock, 1380 @NonNull String operation) { 1381 requireNonNull(lock); 1382 requireNonNull(operation); 1383 1384 try { 1385 lock.lockInterruptibly(); 1386 } catch (InterruptedException e) { 1387 Thread.currentThread().interrupt(); 1388 throw new DatabaseException(format("Interrupted while waiting to %s", operation), e); 1389 } 1390 } 1391 1392 @Nullable 1393 private static Throwable closeStatementContextResources(@NonNull StatementContext<?> statementContext, 1394 @Nullable Throwable cleanupFailure) { 1395 requireNonNull(statementContext); 1396 1397 Queue<AutoCloseable> cleanupOperations = statementContext.getCleanupOperations(); 1398 AutoCloseable cleanupOperation; 1399 1400 while ((cleanupOperation = cleanupOperations.poll()) != null) { 1401 try { 1402 cleanupOperation.close(); 1403 } catch (Throwable cleanupException) { 1404 if (cleanupFailure == null) 1405 cleanupFailure = cleanupException; 1406 else 1407 cleanupFailure.addSuppressed(cleanupException); 1408 } 1409 } 1410 1411 return cleanupFailure; 1412 } 1413 1414 private static boolean isUnsupportedSqlFeature(@NonNull SQLException e) { 1415 requireNonNull(e); 1416 1417 String sqlState = e.getSQLState(); 1418 if (sqlState != null) { 1419 if (sqlState.startsWith("0A") || "HYC00".equals(sqlState)) 1420 return true; 1421 } 1422 1423 Throwable cause = e.getCause(); 1424 if (cause instanceof SQLFeatureNotSupportedException 1425 || cause instanceof UnsupportedOperationException 1426 || cause instanceof AbstractMethodError) 1427 return true; 1428 1429 String message = e.getMessage(); 1430 if (message == null) 1431 return false; 1432 1433 String lower = message.toLowerCase(Locale.ROOT); 1434 return lower.contains("not supported") 1435 || lower.contains("unsupported") 1436 || lower.contains("not implemented") 1437 || lower.contains("feature not supported"); 1438 } 1439 1440 @Nullable 1441 private static Duration validateQueryTimeout(@Nullable Duration queryTimeout) { 1442 if (queryTimeout != null) { 1443 if (queryTimeout.isNegative()) 1444 throw new IllegalArgumentException("queryTimeout must be >= 0"); 1445 1446 queryTimeoutSeconds(queryTimeout); 1447 } 1448 1449 return queryTimeout; 1450 } 1451 1452 @Nullable 1453 private static Integer validateNonNegativeStatementSetting(@NonNull String name, 1454 @Nullable Integer value) { 1455 requireNonNull(name); 1456 1457 if (value != null && value < 0) 1458 throw new IllegalArgumentException(format("%s must be >= 0", name)); 1459 1460 return value; 1461 } 1462 1463 @Nullable 1464 private static Integer validatePositiveQuerySetting(@NonNull String name, 1465 @Nullable Integer value) { 1466 requireNonNull(name); 1467 1468 if (value != null && value <= 0) 1469 throw new IllegalArgumentException(format("%s must be > 0", name)); 1470 1471 return value; 1472 } 1473 1474 private static int queryTimeoutSeconds(@NonNull Duration queryTimeout) { 1475 requireNonNull(queryTimeout); 1476 1477 long seconds = queryTimeout.getSeconds(); 1478 1479 if (queryTimeout.getNano() > 0) { 1480 if (seconds == Long.MAX_VALUE) 1481 throw new IllegalArgumentException(format("queryTimeout must be <= %s seconds", Integer.MAX_VALUE)); 1482 1483 ++seconds; 1484 } 1485 1486 if (seconds > Integer.MAX_VALUE) 1487 throw new IllegalArgumentException(format("queryTimeout must be <= %s seconds", Integer.MAX_VALUE)); 1488 1489 return (int) seconds; 1490 } 1491 1492 private static int healthCheckTimeoutSeconds(@NonNull Duration timeout) { 1493 requireNonNull(timeout); 1494 1495 if (timeout.isNegative()) 1496 throw new IllegalArgumentException("timeout must be >= 0"); 1497 1498 long seconds = timeout.getSeconds(); 1499 1500 if (timeout.getNano() > 0) { 1501 if (seconds == Long.MAX_VALUE) 1502 throw new IllegalArgumentException(format("timeout must be <= %s seconds", Integer.MAX_VALUE)); 1503 1504 ++seconds; 1505 } 1506 1507 if (seconds > Integer.MAX_VALUE) 1508 throw new IllegalArgumentException(format("timeout must be <= %s seconds", Integer.MAX_VALUE)); 1509 1510 return (int) seconds; 1511 } 1512 1513 @NonNull 1514 private ParsedSqlVariants getParsedSqlVariants(@NonNull String sql) { 1515 requireNonNull(sql); 1516 1517 if (this.parsedSqlCache == null) 1518 return parseNamedParameterSqlVariants(sql); 1519 1520 return this.parsedSqlCache.computeIfAbsent(sql, Database::parseNamedParameterSqlVariants); 1521 } 1522 1523 @NonNull 1524 private static ParsedSqlVariants parseNamedParameterSqlVariants(@NonNull String sql) { 1525 requireNonNull(sql); 1526 1527 ParsedSql standardParsedSql = null; 1528 ParsedSql mysqlParsedSql = null; 1529 IllegalArgumentException standardParseFailure = null; 1530 IllegalArgumentException mysqlParseFailure = null; 1531 1532 try { 1533 standardParsedSql = parseNamedParameterSql(sql, SqlLexicalMode.STANDARD); 1534 } catch (IllegalArgumentException e) { 1535 standardParseFailure = e; 1536 } 1537 1538 if (requiresMySqlLexicalVariant(sql)) { 1539 try { 1540 mysqlParsedSql = parseNamedParameterSql(sql, SqlLexicalMode.MYSQL); 1541 } catch (IllegalArgumentException e) { 1542 mysqlParseFailure = e; 1543 } 1544 } else { 1545 mysqlParsedSql = standardParsedSql; 1546 mysqlParseFailure = standardParseFailure; 1547 } 1548 1549 if (standardParsedSql == null && mysqlParsedSql == null) 1550 throw requireNonNull(standardParseFailure); 1551 1552 ParsedSql effectiveStandardParsedSql = standardParsedSql == null 1553 ? requireNonNull(mysqlParsedSql) 1554 : standardParsedSql; 1555 ParsedSql effectiveMysqlParsedSql = mysqlParsedSql == null 1556 ? effectiveStandardParsedSql 1557 : mysqlParsedSql; 1558 Set<String> distinctParameterNames; 1559 1560 if (effectiveStandardParsedSql.distinctParameterNames.equals(effectiveMysqlParsedSql.distinctParameterNames)) { 1561 distinctParameterNames = effectiveStandardParsedSql.distinctParameterNames; 1562 } else { 1563 Set<String> combinedParameterNames = new HashSet<>(effectiveStandardParsedSql.distinctParameterNames); 1564 combinedParameterNames.addAll(effectiveMysqlParsedSql.distinctParameterNames); 1565 distinctParameterNames = Set.copyOf(combinedParameterNames); 1566 } 1567 1568 boolean requiresDatabaseType = standardParseFailure != null 1569 || mysqlParseFailure != null 1570 || !effectiveStandardParsedSql.equivalentTo(effectiveMysqlParsedSql); 1571 1572 return new ParsedSqlVariants(effectiveStandardParsedSql, 1573 effectiveMysqlParsedSql, 1574 standardParseFailure == null ? null : requireNonNull(standardParseFailure.getMessage()), 1575 mysqlParseFailure == null ? null : requireNonNull(mysqlParseFailure.getMessage()), 1576 distinctParameterNames, 1577 requiresDatabaseType); 1578 } 1579 1580 private static boolean requiresMySqlLexicalVariant(@NonNull String sql) { 1581 requireNonNull(sql); 1582 return sql.indexOf('#') >= 0 || sql.indexOf("\\'") >= 0; 1583 } 1584 1585 private static final class ParsedSqlVariants { 1586 @NonNull 1587 private final ParsedSql standardParsedSql; 1588 @NonNull 1589 private final ParsedSql mysqlParsedSql; 1590 @Nullable 1591 private final String standardParseFailureMessage; 1592 @Nullable 1593 private final String mysqlParseFailureMessage; 1594 @NonNull 1595 private final Set<String> distinctParameterNames; 1596 private final boolean requiresDatabaseType; 1597 1598 private ParsedSqlVariants(@NonNull ParsedSql standardParsedSql, 1599 @NonNull ParsedSql mysqlParsedSql, 1600 @Nullable String standardParseFailureMessage, 1601 @Nullable String mysqlParseFailureMessage, 1602 @NonNull Set<String> distinctParameterNames, 1603 boolean requiresDatabaseType) { 1604 this.standardParsedSql = requireNonNull(standardParsedSql); 1605 this.mysqlParsedSql = requireNonNull(mysqlParsedSql); 1606 this.standardParseFailureMessage = standardParseFailureMessage; 1607 this.mysqlParseFailureMessage = mysqlParseFailureMessage; 1608 this.distinctParameterNames = Set.copyOf(requireNonNull(distinctParameterNames)); 1609 this.requiresDatabaseType = requiresDatabaseType; 1610 } 1611 } 1612 1613 /** 1614 * Default internal implementation of {@link Query}. 1615 * <p> 1616 * This class is intended for use by a single thread. 1617 */ 1618 @NotThreadSafe 1619 private static final class DefaultQuery implements Query { 1620 @NonNull 1621 private final Database database; 1622 @NonNull 1623 private final String originalSql; 1624 @NonNull 1625 private final ParsedSqlVariants parsedSqlVariants; 1626 @NonNull 1627 private final Map<String, Object> bindings; 1628 @Nullable 1629 private PreparedStatementCustomizer preparedStatementCustomizer; 1630 @Nullable 1631 private Duration queryTimeout; 1632 @Nullable 1633 private Integer fetchSize; 1634 @Nullable 1635 private Integer maxRows; 1636 @Nullable 1637 private Integer batchChunkSize; 1638 @Nullable 1639 private Object id; 1640 @Nullable 1641 private ResultSetMapper resultSetMapper; 1642 @Nullable 1643 private PreparedStatementBinder preparedStatementBinder; 1644 1645 private DefaultQuery(@NonNull Database database, 1646 @NonNull String sql) { 1647 requireNonNull(database); 1648 requireNonNull(sql); 1649 1650 this.database = database; 1651 this.originalSql = sql; 1652 this.parsedSqlVariants = database.getParsedSqlVariants(sql); 1653 1654 this.bindings = new LinkedHashMap<>(Math.max(8, this.parsedSqlVariants.distinctParameterNames.size())); 1655 this.preparedStatementCustomizer = null; 1656 this.queryTimeout = null; 1657 this.fetchSize = null; 1658 this.maxRows = null; 1659 this.batchChunkSize = null; 1660 } 1661 1662 @NonNull 1663 @Override 1664 public Query bind(@NonNull String name, 1665 @Nullable Object value) { 1666 requireNonNull(name); 1667 1668 if (!this.parsedSqlVariants.distinctParameterNames.contains(name)) 1669 throw new IllegalArgumentException(format("Unknown named parameter '%s' for SQL: %s", name, this.originalSql)); 1670 1671 this.bindings.put(name, value); 1672 return this; 1673 } 1674 1675 @NonNull 1676 @Override 1677 public Query bindAll(@NonNull Map<@NonNull String, @Nullable Object> parameters) { 1678 requireNonNull(parameters); 1679 1680 for (Map.Entry<@NonNull String, @Nullable Object> entry : parameters.entrySet()) 1681 bind(entry.getKey(), entry.getValue()); 1682 1683 return this; 1684 } 1685 1686 @NonNull 1687 @Override 1688 public Query id(@Nullable Object id) { 1689 this.id = id; 1690 return this; 1691 } 1692 1693 @NonNull 1694 @Override 1695 public Query queryTimeout(@Nullable Duration queryTimeout) { 1696 this.queryTimeout = validateQueryTimeout(queryTimeout); 1697 return this; 1698 } 1699 1700 @NonNull 1701 @Override 1702 public Query fetchSize(@Nullable Integer fetchSize) { 1703 this.fetchSize = validateNonNegativeStatementSetting("fetchSize", fetchSize); 1704 return this; 1705 } 1706 1707 @NonNull 1708 @Override 1709 public Query maxRows(@Nullable Integer maxRows) { 1710 this.maxRows = validateNonNegativeStatementSetting("maxRows", maxRows); 1711 return this; 1712 } 1713 1714 @NonNull 1715 @Override 1716 public Query batchChunkSize(@Nullable Integer batchChunkSize) { 1717 this.batchChunkSize = validatePositiveQuerySetting("batchChunkSize", batchChunkSize); 1718 return this; 1719 } 1720 1721 @NonNull 1722 @Override 1723 public Query resultSetMapper(@Nullable ResultSetMapper resultSetMapper) { 1724 this.resultSetMapper = resultSetMapper; 1725 return this; 1726 } 1727 1728 @NonNull 1729 @Override 1730 public Query preparedStatementBinder(@Nullable PreparedStatementBinder preparedStatementBinder) { 1731 this.preparedStatementBinder = preparedStatementBinder; 1732 return this; 1733 } 1734 1735 private StatementContext.@Nullable SpiOverrides spiOverrides() { 1736 return this.resultSetMapper == null && this.preparedStatementBinder == null 1737 ? null 1738 : new StatementContext.SpiOverrides(this.resultSetMapper, this.preparedStatementBinder); 1739 } 1740 1741 @NonNull 1742 @Override 1743 public Query customize(@NonNull PreparedStatementCustomizer preparedStatementCustomizer) { 1744 requireNonNull(preparedStatementCustomizer); 1745 this.preparedStatementCustomizer = preparedStatementCustomizer; 1746 1747 return this; 1748 } 1749 1750 @NonNull 1751 @Override 1752 public <T> Optional<T> fetchObject(@NonNull Class<T> resultType) { 1753 validateBatchChunkSizeNotSet(); 1754 requireNonNull(resultType); 1755 PreparedQuery preparedQuery = prepare(this.bindings); 1756 return this.database.queryForObject(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters); 1757 } 1758 1759 @NonNull 1760 @Override 1761 public <T> List<@Nullable T> fetchList(@NonNull Class<T> resultType) { 1762 validateBatchChunkSizeNotSet(); 1763 requireNonNull(resultType); 1764 PreparedQuery preparedQuery = prepare(this.bindings); 1765 return this.database.queryForList(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters); 1766 } 1767 1768 @Nullable 1769 @Override 1770 public <T, R> R fetchStream(@NonNull Class<T> resultType, 1771 @NonNull Function<Stream<@Nullable T>, R> streamFunction) { 1772 validateBatchChunkSizeNotSet(); 1773 requireNonNull(resultType); 1774 requireNonNull(streamFunction); 1775 PreparedQuery preparedQuery = prepare(this.bindings); 1776 return this.database.queryForStream(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), 1777 this.fetchSize != null, streamFunction, spiOverrides(), preparedQuery.parameters); 1778 } 1779 1780 1781 @NonNull 1782 @Override 1783 public Long execute() { 1784 validateBatchChunkSizeNotSet(); 1785 PreparedQuery preparedQuery = prepare(this.bindings); 1786 return this.database.execute(preparedQuery.statement, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters); 1787 } 1788 1789 @NonNull 1790 @Override 1791 public <T> Optional<T> executeReturningGeneratedKey(@NonNull Class<T> resultType) { 1792 validateBatchChunkSizeNotSet(); 1793 requireNonNull(resultType); 1794 PreparedQuery preparedQuery = prepare(this.bindings); 1795 return this.database.executeReturningGeneratedKey(preparedQuery.statement, resultType, 1796 effectivePreparedStatementCustomizer(), new String[0], spiOverrides(), preparedQuery.parameters); 1797 } 1798 1799 @NonNull 1800 @Override 1801 public <T> Optional<T> executeReturningGeneratedKey(@NonNull Class<T> resultType, 1802 @NonNull String @NonNull ... keyColumnNames) { 1803 validateBatchChunkSizeNotSet(); 1804 requireNonNull(resultType); 1805 PreparedQuery preparedQuery = prepare(this.bindings); 1806 return this.database.executeReturningGeneratedKey(preparedQuery.statement, resultType, 1807 effectivePreparedStatementCustomizer(), keyColumnNames, spiOverrides(), preparedQuery.parameters); 1808 } 1809 1810 @NonNull 1811 @Override 1812 public <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Class<T> resultType) { 1813 validateBatchChunkSizeNotSet(); 1814 requireNonNull(resultType); 1815 PreparedQuery preparedQuery = prepare(this.bindings); 1816 return this.database.executeReturningGeneratedKeys(preparedQuery.statement, resultType, 1817 effectivePreparedStatementCustomizer(), new String[0], spiOverrides(), preparedQuery.parameters); 1818 } 1819 1820 @NonNull 1821 @Override 1822 public <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Class<T> resultType, 1823 @NonNull String @NonNull ... keyColumnNames) { 1824 validateBatchChunkSizeNotSet(); 1825 requireNonNull(resultType); 1826 PreparedQuery preparedQuery = prepare(this.bindings); 1827 return this.database.executeReturningGeneratedKeys(preparedQuery.statement, resultType, 1828 effectivePreparedStatementCustomizer(), keyColumnNames, spiOverrides(), preparedQuery.parameters); 1829 } 1830 1831 @NonNull 1832 @Override 1833 public List<Long> executeBatch(@NonNull List<@NonNull Map<@NonNull String, @Nullable Object>> parameterGroups) { 1834 requireNonNull(parameterGroups); 1835 if (parameterGroups.isEmpty()) 1836 return List.of(); 1837 1838 List<List<Object>> parametersAsList = new ArrayList<>(parameterGroups.size()); 1839 Object statementId = this.id == null ? this.database.generateId() : this.id; 1840 Statement statement = null; 1841 String expandedSql = null; 1842 1843 for (Map<@NonNull String, @Nullable Object> parameterGroup : parameterGroups) { 1844 requireNonNull(parameterGroup); 1845 1846 for (String parameterName : parameterGroup.keySet()) 1847 if (!this.parsedSqlVariants.distinctParameterNames.contains(parameterName)) 1848 throw new IllegalArgumentException(format("Unknown named parameter '%s' for SQL: %s", parameterName, this.originalSql)); 1849 1850 Map<String, Object> mergedBindings; 1851 if (this.bindings.isEmpty()) { 1852 mergedBindings = parameterGroup; 1853 } else if (parameterGroup.isEmpty()) { 1854 mergedBindings = this.bindings; 1855 } else { 1856 Map<String, Object> combinedBindings = new LinkedHashMap<>(this.bindings); 1857 combinedBindings.putAll(parameterGroup); 1858 mergedBindings = combinedBindings; 1859 } 1860 1861 PreparedQuery preparedQuery = prepare(mergedBindings, statementId); 1862 1863 if (expandedSql == null) { 1864 expandedSql = preparedQuery.statement.getSql(); 1865 statement = preparedQuery.statement; 1866 } else if (!expandedSql.equals(preparedQuery.statement.getSql())) { 1867 throw new IllegalArgumentException(format( 1868 "Inconsistent SQL after expanding parameters for batch execution; ensure collection sizes are consistent. SQL: %s", 1869 this.originalSql)); 1870 } 1871 1872 parametersAsList.add(Arrays.asList(preparedQuery.parameters)); 1873 } 1874 1875 if (statement == null) 1876 statement = Statement.of(statementId, buildPlaceholderSql()); 1877 1878 return this.database.executeBatch(statement, parametersAsList, effectivePreparedStatementCustomizer(), this.batchChunkSize, spiOverrides()); 1879 } 1880 1881 @NonNull 1882 @Override 1883 public <T> Optional<T> executeForObject(@NonNull Class<T> resultType) { 1884 validateBatchChunkSizeNotSet(); 1885 requireNonNull(resultType); 1886 PreparedQuery preparedQuery = prepare(this.bindings); 1887 return this.database.executeForObject(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters); 1888 } 1889 1890 @NonNull 1891 @Override 1892 public <T> List<@Nullable T> executeForList(@NonNull Class<T> resultType) { 1893 validateBatchChunkSizeNotSet(); 1894 requireNonNull(resultType); 1895 PreparedQuery preparedQuery = prepare(this.bindings); 1896 return this.database.executeForList(preparedQuery.statement, resultType, effectivePreparedStatementCustomizer(), spiOverrides(), preparedQuery.parameters); 1897 } 1898 1899 private void validateBatchChunkSizeNotSet() { 1900 if (this.batchChunkSize != null) 1901 throw new IllegalStateException("batchChunkSize applies only to executeBatch(...)"); 1902 } 1903 1904 @Nullable 1905 private PreparedStatementCustomizer effectivePreparedStatementCustomizer() { 1906 if (!this.database.hasDefaultPreparedStatementSettings() 1907 && !hasQueryPreparedStatementSettings() 1908 && this.preparedStatementCustomizer == null) 1909 return null; 1910 1911 return (statementContext, preparedStatement) -> { 1912 this.database.applyDefaultPreparedStatementSettings(preparedStatement); 1913 applyPreparedStatementSettings(preparedStatement, this.queryTimeout, this.fetchSize, this.maxRows); 1914 1915 if (this.preparedStatementCustomizer != null) 1916 this.preparedStatementCustomizer.customize(statementContext, preparedStatement); 1917 }; 1918 } 1919 1920 private boolean hasQueryPreparedStatementSettings() { 1921 return this.queryTimeout != null || this.fetchSize != null || this.maxRows != null; 1922 } 1923 1924 @NonNull 1925 private PreparedQuery prepare(@NonNull Map<String, Object> bindings) { 1926 Object statementId = this.id == null ? this.database.generateId() : this.id; 1927 return prepare(bindings, statementId); 1928 } 1929 1930 @NonNull 1931 private PreparedQuery prepare(@NonNull Map<String, Object> bindings, 1932 @NonNull Object statementId) { 1933 requireNonNull(bindings); 1934 requireNonNull(statementId); 1935 1936 ParsedSql parsedSql = parsedSqlForDatabaseType(); 1937 List<String> parameterNames = parsedSql.parameterNames; 1938 List<String> sqlFragments = sqlFragmentsForDatabaseType(parsedSql); 1939 validateBindingsForParsedSql(bindings, parsedSql); 1940 1941 if (parameterNames.isEmpty()) 1942 return new PreparedQuery(Statement.of(statementId, sqlFragments.get(0)), new Object[0]); 1943 1944 StringBuilder sql = new StringBuilder(this.originalSql.length() + parameterNames.size() * 2); 1945 List<String> missingParameterNames = null; 1946 List<Object> parameters = new ArrayList<>(parameterNames.size()); 1947 1948 for (int i = 0; i < parameterNames.size(); ++i) { 1949 String parameterName = parameterNames.get(i); 1950 sql.append(sqlFragments.get(i)); 1951 1952 if (!bindings.containsKey(parameterName)) { 1953 if (missingParameterNames == null) 1954 missingParameterNames = new ArrayList<>(); 1955 1956 missingParameterNames.add(parameterName); 1957 sql.append('?'); 1958 continue; 1959 } 1960 1961 SecureParameterSupport.SecureParameterUnwrapResult secureParameterUnwrapResult = 1962 SecureParameterSupport.unwrapSecureAndOptionalParameterWithMetadata(bindings.get(parameterName)); 1963 SecureParameter secureParameter = secureParameterUnwrapResult.secureParameter(); 1964 Object value = secureParameterUnwrapResult.value(); 1965 1966 if (value instanceof InListParameter inListParameter) { 1967 Object[] elements = inListParameter.getElements(); 1968 1969 if (elements.length == 0) 1970 throw new IllegalArgumentException(format("IN-list parameter '%s' for SQL: %s is empty", parameterName, this.originalSql)); 1971 1972 appendPlaceholders(sql, elements.length); 1973 1974 for (int j = 0; j < elements.length; ++j) { 1975 SecureParameterSupport.SecureParameterUnwrapResult elementUnwrapResult = 1976 SecureParameterSupport.unwrapSecureAndOptionalParameterWithMetadata(elements[j]); 1977 Object element = elementUnwrapResult.value(); 1978 1979 if (element == null) 1980 throw new IllegalArgumentException(format( 1981 "IN-list parameter '%s' for SQL: %s contains null element at index %d. " 1982 + "SQL IN does not match NULL values; use an explicit IS NULL predicate instead.", 1983 parameterName, this.originalSql, j)); 1984 1985 SecureParameter effectiveSecureParameter = secureParameter == null 1986 ? elementUnwrapResult.secureParameter() 1987 : secureParameter; 1988 parameters.add(effectiveSecureParameter == null 1989 ? element 1990 : Parameters.secure(element, SecureParameterSupport.maskOf(effectiveSecureParameter))); 1991 } 1992 } else if (value instanceof Collection<?>) { 1993 throw new IllegalArgumentException(format( 1994 "Collection parameter '%s' for SQL: %s must be wrapped with %s.inList(...) or %s.listOf/%s.setOf(...)", 1995 parameterName, this.originalSql, 1996 Parameters.class.getSimpleName(), 1997 Parameters.class.getSimpleName(), Parameters.class.getSimpleName())); 1998 } else if (value != null && value.getClass().isArray() && !(value instanceof byte[])) { 1999 throw new IllegalArgumentException(format( 2000 "Array parameter '%s' for SQL: %s must be wrapped with %s.inList(...), %s.sqlArrayOf(...), or %s.arrayOf(Class, ...)", 2001 parameterName, this.originalSql, 2002 Parameters.class.getSimpleName(), Parameters.class.getSimpleName(), Parameters.class.getSimpleName())); 2003 } else { 2004 sql.append('?'); 2005 parameters.add(secureParameter == null ? value : secureParameter); 2006 } 2007 } 2008 2009 sql.append(sqlFragments.get(sqlFragments.size() - 1)); 2010 2011 if (missingParameterNames != null) 2012 throw new IllegalArgumentException(format("Missing required named parameters %s for SQL: %s", missingParameterNames, this.originalSql)); 2013 2014 return new PreparedQuery(Statement.of(statementId, sql.toString()), parameters.toArray()); 2015 } 2016 2017 @NonNull 2018 private String buildPlaceholderSql() { 2019 ParsedSql parsedSql = parsedSqlForDatabaseType(); 2020 List<String> parameterNames = parsedSql.parameterNames; 2021 List<String> sqlFragments = sqlFragmentsForDatabaseType(parsedSql); 2022 2023 if (parameterNames.isEmpty()) 2024 return sqlFragments.get(0); 2025 2026 StringBuilder sql = new StringBuilder(this.originalSql.length() + parameterNames.size() * 2); 2027 2028 for (int i = 0; i < parameterNames.size(); ++i) 2029 sql.append(sqlFragments.get(i)).append('?'); 2030 2031 sql.append(sqlFragments.get(sqlFragments.size() - 1)); 2032 return sql.toString(); 2033 } 2034 2035 @NonNull 2036 private List<String> sqlFragmentsForDatabaseType(@NonNull ParsedSql parsedSql) { 2037 requireNonNull(parsedSql); 2038 2039 if (!parsedSql.hasQuestionMarkOperators) 2040 return parsedSql.sqlFragments; 2041 2042 return this.database.getDatabaseDialect().sqlFragmentsForOperators( 2043 parsedSql.hasQuestionMarkOperators, 2044 parsedSql.sqlFragments, 2045 parsedSql.questionMarkOperatorFragmentIndexes); 2046 } 2047 2048 @NonNull 2049 private ParsedSql parsedSqlForDatabaseType() { 2050 if (!this.parsedSqlVariants.requiresDatabaseType) 2051 return this.parsedSqlVariants.standardParsedSql; 2052 2053 DatabaseType databaseType; 2054 Optional<Transaction> transaction = this.database.currentTransactionForDatabaseOperation(); 2055 2056 if (transaction.isPresent()) 2057 databaseType = this.database.getDatabaseType(transaction.get().getConnection()); 2058 else 2059 databaseType = this.database.getDatabaseType(); 2060 2061 if (databaseType == DatabaseType.MYSQL || databaseType == DatabaseType.MARIA_DB) { 2062 if (this.parsedSqlVariants.mysqlParseFailureMessage != null) 2063 throw new IllegalArgumentException(this.parsedSqlVariants.mysqlParseFailureMessage); 2064 2065 return this.parsedSqlVariants.mysqlParsedSql; 2066 } 2067 2068 if (this.parsedSqlVariants.standardParseFailureMessage != null) 2069 throw new IllegalArgumentException(this.parsedSqlVariants.standardParseFailureMessage); 2070 2071 return this.parsedSqlVariants.standardParsedSql; 2072 } 2073 2074 private void validateBindingsForParsedSql(@NonNull Map<String, Object> bindings, 2075 @NonNull ParsedSql parsedSql) { 2076 requireNonNull(bindings); 2077 requireNonNull(parsedSql); 2078 2079 for (String bindingName : bindings.keySet()) 2080 if (!parsedSql.distinctParameterNames.contains(bindingName)) 2081 throw new IllegalArgumentException(format( 2082 "Named parameter '%s' is not valid for the detected database SQL syntax. SQL: %s", 2083 bindingName, this.originalSql)); 2084 } 2085 2086 private void appendPlaceholders(@NonNull StringBuilder sql, 2087 int count) { 2088 requireNonNull(sql); 2089 2090 for (int i = 0; i < count; ++i) { 2091 if (i > 0) 2092 sql.append(", "); 2093 sql.append('?'); 2094 } 2095 } 2096 2097 private static final class PreparedQuery { 2098 @NonNull 2099 private final Statement statement; 2100 @NonNull 2101 private final Object @NonNull [] parameters; 2102 2103 private PreparedQuery(@NonNull Statement statement, 2104 Object @NonNull [] parameters) { 2105 this.statement = requireNonNull(statement); 2106 this.parameters = requireNonNull(parameters); 2107 } 2108 } 2109 2110 } 2111 2112 static final class ParsedSql { 2113 @NonNull 2114 private final List<String> sqlFragments; 2115 @NonNull 2116 private final List<@NonNull List<@NonNull Integer>> questionMarkOperatorFragmentIndexes; 2117 private final boolean hasQuestionMarkOperators; 2118 @NonNull 2119 private final List<String> parameterNames; 2120 @NonNull 2121 private final Set<String> distinctParameterNames; 2122 2123 private ParsedSql(@NonNull List<String> sqlFragments, 2124 @NonNull List<@NonNull List<@NonNull Integer>> questionMarkOperatorFragmentIndexes, 2125 @NonNull List<String> parameterNames, 2126 @NonNull Set<String> distinctParameterNames) { 2127 requireNonNull(sqlFragments); 2128 requireNonNull(questionMarkOperatorFragmentIndexes); 2129 requireNonNull(parameterNames); 2130 requireNonNull(distinctParameterNames); 2131 2132 this.sqlFragments = sqlFragments; 2133 this.questionMarkOperatorFragmentIndexes = questionMarkOperatorFragmentIndexes.stream() 2134 .map(List::copyOf) 2135 .toList(); 2136 this.hasQuestionMarkOperators = questionMarkOperatorFragmentIndexes.stream().anyMatch(indexes -> !indexes.isEmpty()); 2137 this.parameterNames = parameterNames; 2138 this.distinctParameterNames = distinctParameterNames; 2139 } 2140 2141 private boolean equivalentTo(@NonNull ParsedSql other) { 2142 requireNonNull(other); 2143 return this.sqlFragments.equals(other.sqlFragments) 2144 && this.questionMarkOperatorFragmentIndexes.equals(other.questionMarkOperatorFragmentIndexes) 2145 && this.parameterNames.equals(other.parameterNames); 2146 } 2147 } 2148 2149 enum SqlLexicalMode { 2150 STANDARD, 2151 MYSQL 2152 } 2153 2154 @NonNull 2155 static ParsedSql parseNamedParameterSql(@NonNull String sql) { 2156 return parseNamedParameterSql(sql, SqlLexicalMode.STANDARD); 2157 } 2158 2159 @NonNull 2160 static ParsedSql parseNamedParameterSql(@NonNull String sql, 2161 @NonNull SqlLexicalMode lexicalMode) { 2162 requireNonNull(sql); 2163 requireNonNull(lexicalMode); 2164 2165 List<String> sqlFragments = new ArrayList<>(); 2166 StringBuilder sqlFragment = new StringBuilder(sql.length()); 2167 List<List<Integer>> questionMarkOperatorFragmentIndexes = new ArrayList<>(); 2168 List<Integer> currentQuestionMarkOperatorIndexes = new ArrayList<>(); 2169 List<String> parameterNames = new ArrayList<>(); 2170 Set<String> distinctParameterNames = new HashSet<>(); 2171 2172 boolean inSingleQuote = false; 2173 boolean inSingleQuoteEscapesBackslash = false; 2174 int singleQuoteStartIndex = -1; 2175 boolean inDoubleQuote = false; 2176 int doubleQuoteStartIndex = -1; 2177 boolean inBacktickQuote = false; 2178 int backtickQuoteStartIndex = -1; 2179 boolean inBracketQuote = false; 2180 int bracketQuoteStartIndex = -1; 2181 boolean inLineComment = false; 2182 int blockCommentDepth = 0; 2183 int blockCommentStartIndex = -1; 2184 String dollarQuoteDelimiter = null; 2185 int dollarQuoteStartIndex = -1; 2186 Character oracleQuoteClosingDelimiter = null; 2187 int oracleQuoteStartIndex = -1; 2188 int previousMeaningfulIndex = -1; 2189 2190 for (int i = 0; i < sql.length(); ) { 2191 if (oracleQuoteClosingDelimiter != null) { 2192 char c = sql.charAt(i); 2193 sqlFragment.append(c); 2194 2195 if (c == oracleQuoteClosingDelimiter && i + 1 < sql.length() && sql.charAt(i + 1) == '\'') { 2196 sqlFragment.append('\''); 2197 i += 2; 2198 oracleQuoteClosingDelimiter = null; 2199 oracleQuoteStartIndex = -1; 2200 previousMeaningfulIndex = i - 1; 2201 } else { 2202 ++i; 2203 } 2204 2205 continue; 2206 } 2207 2208 if (dollarQuoteDelimiter != null) { 2209 if (sql.startsWith(dollarQuoteDelimiter, i)) { 2210 sqlFragment.append(dollarQuoteDelimiter); 2211 previousMeaningfulIndex = i + dollarQuoteDelimiter.length() - 1; 2212 i += dollarQuoteDelimiter.length(); 2213 dollarQuoteDelimiter = null; 2214 dollarQuoteStartIndex = -1; 2215 } else { 2216 sqlFragment.append(sql.charAt(i)); 2217 ++i; 2218 } 2219 2220 continue; 2221 } 2222 2223 char c = sql.charAt(i); 2224 2225 if (inLineComment) { 2226 sqlFragment.append(c); 2227 ++i; 2228 2229 if (c == '\n' || c == '\r') 2230 inLineComment = false; 2231 2232 continue; 2233 } 2234 2235 if (blockCommentDepth > 0) { 2236 if (c == '/' && i + 1 < sql.length() && sql.charAt(i + 1) == '*') { 2237 sqlFragment.append("/*"); 2238 i += 2; 2239 ++blockCommentDepth; 2240 } else if (c == '*' && i + 1 < sql.length() && sql.charAt(i + 1) == '/') { 2241 sqlFragment.append("*/"); 2242 i += 2; 2243 --blockCommentDepth; 2244 if (blockCommentDepth == 0) 2245 blockCommentStartIndex = -1; 2246 } else { 2247 sqlFragment.append(c); 2248 ++i; 2249 } 2250 2251 continue; 2252 } 2253 2254 if (inSingleQuote) { 2255 sqlFragment.append(c); 2256 2257 if (inSingleQuoteEscapesBackslash && c == '\\' && i + 1 < sql.length()) { 2258 sqlFragment.append(sql.charAt(i + 1)); 2259 i += 2; 2260 continue; 2261 } 2262 2263 if (c == '\'') { 2264 // Escaped quote: '' 2265 if (i + 1 < sql.length() && sql.charAt(i + 1) == '\'') { 2266 sqlFragment.append('\''); 2267 i += 2; 2268 continue; 2269 } 2270 2271 inSingleQuote = false; 2272 inSingleQuoteEscapesBackslash = false; 2273 singleQuoteStartIndex = -1; 2274 previousMeaningfulIndex = i; 2275 } 2276 2277 ++i; 2278 continue; 2279 } 2280 2281 if (inDoubleQuote) { 2282 sqlFragment.append(c); 2283 2284 if (c == '"') { 2285 // Escaped quote: "" 2286 if (i + 1 < sql.length() && sql.charAt(i + 1) == '"') { 2287 sqlFragment.append('"'); 2288 i += 2; 2289 continue; 2290 } 2291 2292 inDoubleQuote = false; 2293 doubleQuoteStartIndex = -1; 2294 previousMeaningfulIndex = i; 2295 } 2296 2297 ++i; 2298 continue; 2299 } 2300 2301 if (inBacktickQuote) { 2302 sqlFragment.append(c); 2303 2304 if (c == '`') { 2305 inBacktickQuote = false; 2306 backtickQuoteStartIndex = -1; 2307 previousMeaningfulIndex = i; 2308 } 2309 2310 ++i; 2311 continue; 2312 } 2313 2314 if (inBracketQuote) { 2315 sqlFragment.append(c); 2316 2317 if (c == ']' && i + 1 < sql.length() && sql.charAt(i + 1) == ']') { 2318 sqlFragment.append(']'); 2319 i += 2; 2320 continue; 2321 } 2322 2323 if (c == ']') { 2324 inBracketQuote = false; 2325 bracketQuoteStartIndex = -1; 2326 previousMeaningfulIndex = i; 2327 } 2328 2329 ++i; 2330 continue; 2331 } 2332 2333 // Not inside string/comment 2334 if (lexicalMode == SqlLexicalMode.MYSQL && c == '#') { 2335 sqlFragment.append(c); 2336 ++i; 2337 inLineComment = true; 2338 continue; 2339 } 2340 2341 if (c == '-' && i + 1 < sql.length() && sql.charAt(i + 1) == '-') { 2342 sqlFragment.append("--"); 2343 i += 2; 2344 inLineComment = true; 2345 continue; 2346 } 2347 2348 if (c == '/' && i + 1 < sql.length() && sql.charAt(i + 1) == '*') { 2349 sqlFragment.append("/*"); 2350 i += 2; 2351 blockCommentDepth = 1; 2352 blockCommentStartIndex = i - 2; 2353 continue; 2354 } 2355 2356 if ((c == 'U' || c == 'u') && !isIdentifierContinuation(sql, i) 2357 && i + 2 < sql.length() && sql.charAt(i + 1) == '&' && sql.charAt(i + 2) == '\'') { 2358 inSingleQuote = true; 2359 inSingleQuoteEscapesBackslash = true; 2360 singleQuoteStartIndex = i; 2361 sqlFragment.append(c).append("&'"); 2362 i += 3; 2363 continue; 2364 } 2365 2366 if ((c == 'E' || c == 'e') && !isIdentifierContinuation(sql, i) 2367 && i + 1 < sql.length() && sql.charAt(i + 1) == '\'') { 2368 inSingleQuote = true; 2369 inSingleQuoteEscapesBackslash = true; 2370 singleQuoteStartIndex = i; 2371 sqlFragment.append(c).append('\''); 2372 i += 2; 2373 continue; 2374 } 2375 2376 if ((c == 'Q' || c == 'q') && !isIdentifierContinuation(sql, i) 2377 && i + 2 < sql.length() && sql.charAt(i + 1) == '\'' 2378 && !Character.isWhitespace(sql.charAt(i + 2)) && sql.charAt(i + 2) != '\'') { 2379 char openingDelimiter = sql.charAt(i + 2); 2380 oracleQuoteClosingDelimiter = oracleQuoteClosingDelimiter(openingDelimiter); 2381 oracleQuoteStartIndex = i; 2382 sqlFragment.append(c).append('\'').append(openingDelimiter); 2383 i += 3; 2384 continue; 2385 } 2386 2387 if (c == '\'') { 2388 inSingleQuote = true; 2389 inSingleQuoteEscapesBackslash = lexicalMode == SqlLexicalMode.MYSQL; 2390 singleQuoteStartIndex = i; 2391 sqlFragment.append(c); 2392 ++i; 2393 continue; 2394 } 2395 2396 if (c == '"') { 2397 inDoubleQuote = true; 2398 doubleQuoteStartIndex = i; 2399 sqlFragment.append(c); 2400 ++i; 2401 continue; 2402 } 2403 2404 if (c == '`') { 2405 inBacktickQuote = true; 2406 backtickQuoteStartIndex = i; 2407 sqlFragment.append(c); 2408 ++i; 2409 continue; 2410 } 2411 2412 if (c == '[' && !isBracketSubscriptStart(sql, i)) { 2413 inBracketQuote = true; 2414 bracketQuoteStartIndex = i; 2415 sqlFragment.append(c); 2416 ++i; 2417 continue; 2418 } 2419 2420 if (c == '$' && !isIdentifierContinuation(sql, i)) { 2421 String delimiter = parseDollarQuoteDelimiter(sql, i); 2422 2423 if (delimiter != null) { 2424 sqlFragment.append(delimiter); 2425 i += delimiter.length(); 2426 dollarQuoteDelimiter = delimiter; 2427 dollarQuoteStartIndex = i - delimiter.length(); 2428 continue; 2429 } 2430 } 2431 2432 if (c == '?') { 2433 if (isAllowedQuestionMarkOperator(sql, i, previousMeaningfulIndex)) { 2434 currentQuestionMarkOperatorIndexes.add(sqlFragment.length()); 2435 sqlFragment.append(c); 2436 ++i; 2437 continue; 2438 } 2439 2440 throw new IllegalArgumentException(format("Positional parameters ('?') are not supported. Use named parameters (e.g. ':id') and %s#bind. SQL: %s", 2441 Query.class.getSimpleName(), sql)); 2442 } 2443 2444 if (c == ':' && i + 1 < sql.length() && sql.charAt(i + 1) == ':') { 2445 // Postgres type-cast operator (::), do not treat second ':' as a parameter prefix. 2446 sqlFragment.append("::"); 2447 i += 2; 2448 continue; 2449 } 2450 2451 if (c == ':' && i + 1 < sql.length() && Character.isJavaIdentifierStart(sql.charAt(i + 1))) { 2452 int nameStartIndex = i + 1; 2453 int nameEndIndex = nameStartIndex + 1; 2454 2455 while (nameEndIndex < sql.length() && Character.isJavaIdentifierPart(sql.charAt(nameEndIndex))) 2456 ++nameEndIndex; 2457 2458 String parameterName = sql.substring(nameStartIndex, nameEndIndex); 2459 parameterNames.add(parameterName); 2460 distinctParameterNames.add(parameterName); 2461 sqlFragments.add(sqlFragment.toString()); 2462 questionMarkOperatorFragmentIndexes.add(List.copyOf(currentQuestionMarkOperatorIndexes)); 2463 currentQuestionMarkOperatorIndexes.clear(); 2464 sqlFragment.setLength(0); 2465 i = nameEndIndex; 2466 previousMeaningfulIndex = nameEndIndex - 1; 2467 continue; 2468 } 2469 2470 sqlFragment.append(c); 2471 if (!Character.isWhitespace(c)) 2472 previousMeaningfulIndex = i; 2473 ++i; 2474 } 2475 2476 validateParserTerminalState(sql, inSingleQuote, singleQuoteStartIndex, inDoubleQuote, doubleQuoteStartIndex, 2477 inBacktickQuote, backtickQuoteStartIndex, inBracketQuote, bracketQuoteStartIndex, 2478 blockCommentDepth, blockCommentStartIndex, dollarQuoteDelimiter, dollarQuoteStartIndex, 2479 oracleQuoteClosingDelimiter, oracleQuoteStartIndex); 2480 2481 sqlFragments.add(sqlFragment.toString()); 2482 questionMarkOperatorFragmentIndexes.add(List.copyOf(currentQuestionMarkOperatorIndexes)); 2483 2484 return new ParsedSql(List.copyOf(sqlFragments), List.copyOf(questionMarkOperatorFragmentIndexes), 2485 List.copyOf(parameterNames), Set.copyOf(distinctParameterNames)); 2486 } 2487 2488 private static char oracleQuoteClosingDelimiter(char openingDelimiter) { 2489 return switch (openingDelimiter) { 2490 case '[' -> ']'; 2491 case '{' -> '}'; 2492 case '(' -> ')'; 2493 case '<' -> '>'; 2494 default -> openingDelimiter; 2495 }; 2496 } 2497 2498 @Nullable 2499 private static String parseDollarQuoteDelimiter(@NonNull String sql, 2500 int startIndex) { 2501 requireNonNull(sql); 2502 2503 if (startIndex < 0 || startIndex >= sql.length()) 2504 return null; 2505 2506 if (sql.charAt(startIndex) != '$') 2507 return null; 2508 2509 int i = startIndex + 1; 2510 2511 if (i >= sql.length()) 2512 return null; 2513 2514 char firstTagCharacter = sql.charAt(i); 2515 2516 if (firstTagCharacter == '$') 2517 return "$$"; 2518 2519 if (!isDollarQuoteTagStart(firstTagCharacter)) 2520 return null; 2521 2522 ++i; 2523 2524 while (i < sql.length()) { 2525 char c = sql.charAt(i); 2526 2527 if (c == '$') 2528 return sql.substring(startIndex, i + 1); 2529 2530 if (!isDollarQuoteTagPart(c)) 2531 return null; 2532 2533 ++i; 2534 } 2535 2536 return null; 2537 } 2538 2539 private static boolean isDollarQuoteTagStart(char character) { 2540 return Character.isLetter(character) || character == '_'; 2541 } 2542 2543 private static boolean isDollarQuoteTagPart(char character) { 2544 return Character.isLetterOrDigit(character) || character == '_'; 2545 } 2546 2547 private static boolean isIdentifierContinuation(@NonNull String sql, 2548 int startIndex) { 2549 requireNonNull(sql); 2550 2551 if (startIndex <= 0) 2552 return false; 2553 2554 return Character.isJavaIdentifierPart(sql.charAt(startIndex - 1)); 2555 } 2556 2557 private static boolean isBracketSubscriptStart(@NonNull String sql, 2558 int startIndex) { 2559 requireNonNull(sql); 2560 2561 if (startIndex <= 0) 2562 return false; 2563 2564 int previousIndex = startIndex - 1; 2565 2566 while (previousIndex >= 0 && Character.isWhitespace(sql.charAt(previousIndex))) 2567 --previousIndex; 2568 2569 if (previousIndex < 0) 2570 return false; 2571 2572 char previousChar = sql.charAt(previousIndex); 2573 if (Character.isJavaIdentifierPart(previousChar)) { 2574 int tokenStartIndex = previousIndex; 2575 2576 while (tokenStartIndex > 0 && Character.isJavaIdentifierPart(sql.charAt(tokenStartIndex - 1))) 2577 --tokenStartIndex; 2578 2579 String previousToken = sql.substring(tokenStartIndex, previousIndex + 1).toUpperCase(Locale.ROOT); 2580 return !BRACKET_IDENTIFIER_CONTEXT_KEYWORDS.contains(previousToken); 2581 } 2582 2583 return previousChar == ')' 2584 || previousChar == ']' 2585 || previousChar == '"'; 2586 } 2587 2588 private static void validateParserTerminalState(@NonNull String sql, 2589 boolean inSingleQuote, 2590 int singleQuoteStartIndex, 2591 boolean inDoubleQuote, 2592 int doubleQuoteStartIndex, 2593 boolean inBacktickQuote, 2594 int backtickQuoteStartIndex, 2595 boolean inBracketQuote, 2596 int bracketQuoteStartIndex, 2597 int blockCommentDepth, 2598 int blockCommentStartIndex, 2599 @Nullable String dollarQuoteDelimiter, 2600 int dollarQuoteStartIndex, 2601 @Nullable Character oracleQuoteClosingDelimiter, 2602 int oracleQuoteStartIndex) { 2603 requireNonNull(sql); 2604 2605 if (inSingleQuote) 2606 throw unterminatedSqlConstructException("single-quoted string", singleQuoteStartIndex, sql); 2607 if (inDoubleQuote) 2608 throw unterminatedSqlConstructException("double-quoted identifier", doubleQuoteStartIndex, sql); 2609 if (inBacktickQuote) 2610 throw unterminatedSqlConstructException("backtick-quoted identifier", backtickQuoteStartIndex, sql); 2611 if (inBracketQuote) 2612 throw unterminatedSqlConstructException("bracket-quoted identifier", bracketQuoteStartIndex, sql); 2613 if (blockCommentDepth > 0) 2614 throw unterminatedSqlConstructException("block comment", blockCommentStartIndex, sql); 2615 if (dollarQuoteDelimiter != null) 2616 throw unterminatedSqlConstructException(format("dollar-quoted string %s", dollarQuoteDelimiter), dollarQuoteStartIndex, sql); 2617 if (oracleQuoteClosingDelimiter != null) 2618 throw unterminatedSqlConstructException("Oracle alternative-quoted string", oracleQuoteStartIndex, sql); 2619 } 2620 2621 @NonNull 2622 private static IllegalArgumentException unterminatedSqlConstructException(@NonNull String construct, 2623 int startIndex, 2624 @NonNull String sql) { 2625 requireNonNull(construct); 2626 requireNonNull(sql); 2627 return new IllegalArgumentException(format("Unterminated %s starting at index %s. SQL: %s", construct, startIndex, sql)); 2628 } 2629 2630 @NonNull 2631 private static final Set<@NonNull String> QUESTION_MARK_PREFIX_KEYWORDS = Set.of( 2632 "SELECT", "WHERE", "AND", "OR", "ON", "HAVING", "WHEN", "THEN", "ELSE", "IN", 2633 "VALUES", "SET", "RETURNING", "USING", "LIKE", "BETWEEN", "IS", "NOT", "NULL", 2634 "JOIN", "FROM" 2635 ); 2636 2637 @NonNull 2638 private static final Set<@NonNull String> BRACKET_IDENTIFIER_CONTEXT_KEYWORDS = Set.of( 2639 "SELECT", "FROM", "JOIN", "AS", "INTO", "UPDATE", "TABLE", "BY", "ORDER", "GROUP" 2640 ); 2641 2642 @NonNull 2643 private static final Set<@NonNull String> QUESTION_MARK_SUFFIX_KEYWORDS = Set.of( 2644 "FROM", "WHERE", "AND", "OR", "GROUP", "ORDER", "HAVING", "LIMIT", "OFFSET", 2645 "UNION", "EXCEPT", "INTERSECT", "RETURNING", "JOIN", "ON" 2646 ); 2647 2648 private static boolean isAllowedQuestionMarkOperator(@NonNull String sql, 2649 int questionMarkIndex, 2650 int previousMeaningfulIndex) { 2651 requireNonNull(sql); 2652 2653 int previousIndex = previousMeaningfulIndex; 2654 int nextIndex = nextNonWhitespaceIndex(sql, questionMarkIndex + 1); 2655 2656 if (previousIndex < 0 || nextIndex < 0) 2657 return false; 2658 2659 char previousChar = sql.charAt(previousIndex); 2660 char nextChar = sql.charAt(nextIndex); 2661 2662 if (isOperatorBeforeQuestionMark(previousChar)) 2663 return false; 2664 2665 if (isTerminatorAfterQuestionMark(nextChar)) 2666 return false; 2667 2668 if (questionMarkIndex + 1 < sql.length()) { 2669 char immediateNextChar = sql.charAt(questionMarkIndex + 1); 2670 if (immediateNextChar == '|' || immediateNextChar == '&') { 2671 if (questionMarkIndex + 2 < sql.length() && sql.charAt(questionMarkIndex + 2) == immediateNextChar) 2672 return false; 2673 2674 String previousKeyword = keywordBefore(sql, previousIndex); 2675 if (previousKeyword != null && QUESTION_MARK_PREFIX_KEYWORDS.contains(previousKeyword) 2676 && !isNamedParameterKeywordBefore(sql, previousIndex, previousKeyword)) 2677 return false; 2678 2679 String nextKeyword = keywordAfter(sql, nextIndex); 2680 if (nextKeyword != null && QUESTION_MARK_SUFFIX_KEYWORDS.contains(nextKeyword)) 2681 return false; 2682 2683 return true; 2684 } 2685 } 2686 2687 String previousKeyword = keywordBefore(sql, previousIndex); 2688 if (previousKeyword != null && QUESTION_MARK_PREFIX_KEYWORDS.contains(previousKeyword)) 2689 return false; 2690 2691 String nextKeyword = keywordAfter(sql, nextIndex); 2692 if (nextKeyword != null && QUESTION_MARK_SUFFIX_KEYWORDS.contains(nextKeyword)) 2693 return false; 2694 2695 return true; 2696 } 2697 2698 private static boolean isOperatorBeforeQuestionMark(char c) { 2699 return switch (c) { 2700 case '=', '<', '>', '!', '+', '-', '*', '/', '%', ',', '(' -> true; 2701 default -> false; 2702 }; 2703 } 2704 2705 private static boolean isTerminatorAfterQuestionMark(char c) { 2706 return switch (c) { 2707 case ')', ',', ';' -> true; 2708 default -> false; 2709 }; 2710 } 2711 2712 private static int nextNonWhitespaceIndex(@NonNull String sql, 2713 int startIndex) { 2714 for (int i = startIndex; i < sql.length(); i++) 2715 if (!Character.isWhitespace(sql.charAt(i))) 2716 return i; 2717 return -1; 2718 } 2719 2720 @Nullable 2721 private static String keywordBefore(@NonNull String sql, 2722 int index) { 2723 char c = sql.charAt(index); 2724 if (!Character.isJavaIdentifierPart(c)) 2725 return null; 2726 2727 int endIndex = index + 1; 2728 int startIndex = index; 2729 while (startIndex >= 0 && Character.isJavaIdentifierPart(sql.charAt(startIndex))) 2730 --startIndex; 2731 2732 return sql.substring(startIndex + 1, endIndex).toUpperCase(Locale.ROOT); 2733 } 2734 2735 private static boolean isNamedParameterKeywordBefore(@NonNull String sql, 2736 int keywordEndIndex, 2737 @NonNull String keyword) { 2738 requireNonNull(sql); 2739 requireNonNull(keyword); 2740 2741 int startIndex = keywordEndIndex - keyword.length() + 1; 2742 return startIndex > 0 && sql.charAt(startIndex - 1) == ':'; 2743 } 2744 2745 @Nullable 2746 private static String keywordAfter(@NonNull String sql, 2747 int index) { 2748 char c = sql.charAt(index); 2749 if (!Character.isJavaIdentifierPart(c)) 2750 return null; 2751 2752 int endIndex = index + 1; 2753 while (endIndex < sql.length() && Character.isJavaIdentifierPart(sql.charAt(endIndex))) 2754 ++endIndex; 2755 2756 return sql.substring(index, endIndex).toUpperCase(Locale.ROOT); 2757 } 2758 2759 /** 2760 * Performs a SQL query that is expected to return 0 or 1 result rows. 2761 * 2762 * @param sql the SQL query to execute 2763 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 2764 * @param parameters {@link PreparedStatement} parameters, if any 2765 * @param <T> the type to be returned 2766 * @return a single result (or no result) 2767 * @throws DatabaseException if > 1 row is returned 2768 */ 2769 @NonNull 2770 private <T> Optional<T> queryForObject(@NonNull String sql, 2771 @NonNull Class<T> resultSetRowType, 2772 Object @Nullable ... parameters) { 2773 requireNonNull(sql); 2774 requireNonNull(resultSetRowType); 2775 2776 return queryForObject(Statement.of(generateId(), sql), resultSetRowType, parameters); 2777 } 2778 2779 /** 2780 * Performs a SQL query that is expected to return 0 or 1 result rows. 2781 * 2782 * @param statement the SQL statement to execute 2783 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 2784 * @param parameters {@link PreparedStatement} parameters, if any 2785 * @param <T> the type to be returned 2786 * @return a single result (or no result) 2787 * @throws DatabaseException if > 1 row is returned 2788 */ 2789 private <T> Optional<T> queryForObject(@NonNull Statement statement, 2790 @NonNull Class<T> resultSetRowType, 2791 Object @Nullable ... parameters) { 2792 requireNonNull(statement); 2793 requireNonNull(resultSetRowType); 2794 2795 return queryForObject(statement, resultSetRowType, null, null, parameters); 2796 } 2797 2798 private <T> Optional<T> queryForObject(@NonNull Statement statement, 2799 @NonNull Class<T> resultSetRowType, 2800 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 2801 StatementContext.@Nullable SpiOverrides spiOverrides, 2802 Object @Nullable ... parameters) { 2803 requireNonNull(statement); 2804 requireNonNull(resultSetRowType); 2805 2806 ResultHolder<Optional<T>> resultHolder = new ResultHolder<>(); 2807 StatementContext<T> statementContext = StatementContext.<T>with(statement, this) 2808 .resultSetRowType(resultSetRowType) 2809 .parameters(parameters) 2810 .spiOverrides(spiOverrides) 2811 .build(); 2812 2813 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 2814 2815 performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> { 2816 long startTime = nanoTime(); 2817 2818 try (ResultSet resultSet = preparedStatement.executeQuery()) { 2819 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 2820 startTime = nanoTime(); 2821 2822 Optional<T> result = Optional.empty(); 2823 long rowsReturned = 0L; 2824 2825 if (resultSet.next()) { 2826 rowsReturned = 1L; 2827 try { 2828 T value = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null); 2829 result = Optional.ofNullable(value); 2830 } catch (SQLException e) { 2831 throw databaseExceptionWithStatementContext(statementContext, 2832 format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), statementContext.getResultSetRowType().get()), e); 2833 } 2834 2835 if (resultSet.next()) 2836 throw databaseExceptionWithStatementContext(statementContext, 2837 "Expected 1 row in resultset but got more than 1 instead", 2838 new IllegalStateException("Expected 1 row in resultset but got more than 1 instead")); 2839 } 2840 2841 resultHolder.value = result; 2842 Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime); 2843 StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled() 2844 ? StatementResult.ofRowsReturned(rowsReturned) 2845 : StatementResult.empty(); 2846 return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult); 2847 } 2848 }); 2849 2850 return resultHolder.value; 2851 } 2852 2853 /** 2854 * Performs a SQL query that is expected to return any number of result rows. 2855 * 2856 * @param sql the SQL query to execute 2857 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 2858 * @param parameters {@link PreparedStatement} parameters, if any 2859 * @param <T> the type to be returned 2860 * @return a list of results 2861 */ 2862 @NonNull 2863 private <T> List<@Nullable T> queryForList(@NonNull String sql, 2864 @NonNull Class<T> resultSetRowType, 2865 Object @Nullable ... parameters) { 2866 requireNonNull(sql); 2867 requireNonNull(resultSetRowType); 2868 2869 return queryForList(Statement.of(generateId(), sql), resultSetRowType, parameters); 2870 } 2871 2872 /** 2873 * Performs a SQL query that is expected to return any number of result rows. 2874 * 2875 * @param statement the SQL statement to execute 2876 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 2877 * @param parameters {@link PreparedStatement} parameters, if any 2878 * @param <T> the type to be returned 2879 * @return a list of results 2880 */ 2881 @NonNull 2882 private <T> List<@Nullable T> queryForList(@NonNull Statement statement, 2883 @NonNull Class<T> resultSetRowType, 2884 Object @Nullable ... parameters) { 2885 requireNonNull(statement); 2886 requireNonNull(resultSetRowType); 2887 2888 return queryForList(statement, resultSetRowType, null, null, parameters); 2889 } 2890 2891 private <T> List<@Nullable T> queryForList(@NonNull Statement statement, 2892 @NonNull Class<T> resultSetRowType, 2893 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 2894 StatementContext.@Nullable SpiOverrides spiOverrides, 2895 Object @Nullable ... parameters) { 2896 requireNonNull(statement); 2897 requireNonNull(resultSetRowType); 2898 2899 List<T> list = new ArrayList<>(); 2900 StatementContext<T> statementContext = StatementContext.<T>with(statement, this) 2901 .resultSetRowType(resultSetRowType) 2902 .parameters(parameters) 2903 .spiOverrides(spiOverrides) 2904 .build(); 2905 2906 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 2907 2908 performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> { 2909 long startTime = nanoTime(); 2910 2911 try (ResultSet resultSet = preparedStatement.executeQuery()) { 2912 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 2913 startTime = nanoTime(); 2914 2915 while (resultSet.next()) { 2916 try { 2917 T listElement = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null); 2918 list.add(listElement); 2919 } catch (SQLException e) { 2920 throw databaseExceptionWithStatementContext(statementContext, 2921 format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), statementContext.getResultSetRowType().get()), e); 2922 } 2923 } 2924 2925 Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime); 2926 StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled() 2927 ? StatementResult.ofRowsReturned((long) list.size()) 2928 : StatementResult.empty(); 2929 return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult); 2930 } 2931 }); 2932 2933 return list; 2934 } 2935 2936 @Nullable 2937 private <T, R> R queryForStream(@NonNull Statement statement, 2938 @NonNull Class<T> resultSetRowType, 2939 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 2940 boolean queryFetchSizeConfigured, 2941 @NonNull Function<Stream<@Nullable T>, R> streamFunction, 2942 StatementContext.@Nullable SpiOverrides spiOverrides, 2943 Object @Nullable ... parameters) { 2944 requireNonNull(statement); 2945 requireNonNull(resultSetRowType); 2946 requireNonNull(streamFunction); 2947 2948 StatementContext<T> statementContext = StatementContext.<T>with(statement, this) 2949 .resultSetRowType(resultSetRowType) 2950 .parameters(parameters) 2951 .spiOverrides(spiOverrides) 2952 .build(); 2953 2954 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 2955 StreamingResultSet<T> iterator = new StreamingResultSet<>(this, statementContext, parametersAsList, 2956 preparedStatementCustomizer, queryFetchSizeConfigured); 2957 2958 try { 2959 try (Stream<@Nullable T> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false) 2960 .onClose(iterator::close)) { 2961 try { 2962 return streamFunction.apply(stream); 2963 } catch (Throwable throwable) { 2964 iterator.callbackFailed(throwable); 2965 if (throwable instanceof RuntimeException runtimeException) 2966 throw runtimeException; 2967 if (throwable instanceof Error error) 2968 throw error; 2969 throw new RuntimeException(throwable); 2970 } 2971 } 2972 } finally { 2973 iterator.emitTerminalMetrics(); 2974 } 2975 } 2976 2977 /** 2978 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}; 2979 * or a SQL statement that returns nothing, such as a DDL statement. 2980 * 2981 * @param sql the SQL to execute 2982 * @param parameters {@link PreparedStatement} parameters, if any 2983 * @return the number of rows affected by the SQL statement 2984 */ 2985 @NonNull 2986 private Long execute(@NonNull String sql, 2987 Object @Nullable ... parameters) { 2988 requireNonNull(sql); 2989 return execute(Statement.of(generateId(), sql), parameters); 2990 } 2991 2992 /** 2993 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}; 2994 * or a SQL statement that returns nothing, such as a DDL statement. 2995 * 2996 * @param statement the SQL statement to execute 2997 * @param parameters {@link PreparedStatement} parameters, if any 2998 * @return the number of rows affected by the SQL statement 2999 */ 3000 @NonNull 3001 private Long execute(@NonNull Statement statement, 3002 Object @Nullable ... parameters) { 3003 requireNonNull(statement); 3004 3005 return execute(statement, null, null, parameters); 3006 } 3007 3008 private Long execute(@NonNull Statement statement, 3009 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3010 StatementContext.@Nullable SpiOverrides spiOverrides, 3011 Object @Nullable ... parameters) { 3012 requireNonNull(statement); 3013 3014 ResultHolder<Long> resultHolder = new ResultHolder<>(); 3015 StatementContext<Void> statementContext = StatementContext.with(statement, this) 3016 .parameters(parameters) 3017 .spiOverrides(spiOverrides) 3018 .build(); 3019 3020 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 3021 3022 performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> { 3023 long startTime = nanoTime(); 3024 resultHolder.value = executeUpdate(preparedStatement); 3025 3026 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 3027 StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled() 3028 ? StatementResult.ofRowsAffected(resultHolder.value) 3029 : StatementResult.empty(); 3030 return new DatabaseOperationResult(executionDuration, null, statementResult); 3031 }); 3032 3033 return resultHolder.value; 3034 } 3035 3036 @NonNull 3037 private Long executeUpdate(@NonNull PreparedStatement preparedStatement) throws SQLException { 3038 requireNonNull(preparedStatement); 3039 3040 DatabaseOperationSupportStatus executeLargeUpdateSupported = getExecuteLargeUpdateSupported(); 3041 3042 // Use the appropriate "large" value if we know it. 3043 // If we don't know it, detect it and store it. 3044 if (executeLargeUpdateSupported == DatabaseOperationSupportStatus.YES) 3045 return preparedStatement.executeLargeUpdate(); 3046 3047 if (executeLargeUpdateSupported == DatabaseOperationSupportStatus.NO) 3048 return (long) preparedStatement.executeUpdate(); 3049 3050 // If the driver doesn't support executeLargeUpdate, then UnsupportedOperationException is thrown. 3051 try { 3052 Long result = preparedStatement.executeLargeUpdate(); 3053 setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.YES); 3054 return result; 3055 } catch (SQLFeatureNotSupportedException | UnsupportedOperationException | AbstractMethodError e) { 3056 setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.NO); 3057 return (long) preparedStatement.executeUpdate(); 3058 } catch (SQLException e) { 3059 if (isUnsupportedSqlFeature(e)) { 3060 setExecuteLargeUpdateSupported(DatabaseOperationSupportStatus.NO); 3061 return (long) preparedStatement.executeUpdate(); 3062 } 3063 3064 throw e; 3065 } 3066 } 3067 3068 @NonNull 3069 private <T> Optional<T> executeReturningGeneratedKey(@NonNull Statement statement, 3070 @NonNull Class<T> resultSetRowType, 3071 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3072 @Nullable String @Nullable [] keyColumnNames, 3073 StatementContext.@Nullable SpiOverrides spiOverrides, 3074 Object @Nullable ... parameters) { 3075 requireNonNull(statement); 3076 requireNonNull(resultSetRowType); 3077 3078 ResultHolder<Optional<T>> resultHolder = new ResultHolder<>(); 3079 StatementContext<T> statementContext = StatementContext.<T>with(statement, this) 3080 .resultSetRowType(resultSetRowType) 3081 .parameters(parameters) 3082 .spiOverrides(spiOverrides) 3083 .build(); 3084 3085 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 3086 String[] requestedKeyColumnNames = copyGeneratedKeyColumnNames(keyColumnNames); 3087 3088 performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> { 3089 long startTime = nanoTime(); 3090 Long rowsAffected = executeUpdate(preparedStatement); 3091 3092 try (ResultSet resultSet = preparedStatement.getGeneratedKeys()) { 3093 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 3094 startTime = nanoTime(); 3095 3096 Optional<T> result = Optional.empty(); 3097 long rowsReturned = 0L; 3098 3099 if (resultSet.next()) { 3100 rowsReturned = 1L; 3101 try { 3102 T value = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null); 3103 result = Optional.ofNullable(value); 3104 } catch (SQLException e) { 3105 throw databaseExceptionWithStatementContext(statementContext, 3106 format("Unable to map JDBC generated-key row to %s", statementContext.getResultSetRowType().get()), e); 3107 } 3108 3109 if (resultSet.next()) 3110 throw databaseExceptionWithStatementContext(statementContext, 3111 "Expected 1 generated-key row but got more than 1 instead", 3112 new IllegalStateException("Expected 1 generated-key row but got more than 1 instead")); 3113 } 3114 3115 resultHolder.value = result; 3116 Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime); 3117 StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled() 3118 ? new StatementResult(rowsReturned, rowsAffected) 3119 : StatementResult.empty(); 3120 return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult); 3121 } 3122 }, generatedKeysPreparedStatementFactory(requestedKeyColumnNames)); 3123 3124 return resultHolder.value; 3125 } 3126 3127 @NonNull 3128 private <T> List<@Nullable T> executeReturningGeneratedKeys(@NonNull Statement statement, 3129 @NonNull Class<T> resultSetRowType, 3130 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3131 @Nullable String @Nullable [] keyColumnNames, 3132 StatementContext.@Nullable SpiOverrides spiOverrides, 3133 Object @Nullable ... parameters) { 3134 requireNonNull(statement); 3135 requireNonNull(resultSetRowType); 3136 3137 List<T> list = new ArrayList<>(); 3138 StatementContext<T> statementContext = StatementContext.<T>with(statement, this) 3139 .resultSetRowType(resultSetRowType) 3140 .parameters(parameters) 3141 .spiOverrides(spiOverrides) 3142 .build(); 3143 3144 List<Object> parametersAsList = parameters == null ? List.of() : Arrays.asList(parameters); 3145 String[] requestedKeyColumnNames = copyGeneratedKeyColumnNames(keyColumnNames); 3146 3147 performDatabaseOperation(statementContext, parametersAsList, preparedStatementCustomizer, (PreparedStatement preparedStatement) -> { 3148 long startTime = nanoTime(); 3149 Long rowsAffected = executeUpdate(preparedStatement); 3150 3151 try (ResultSet resultSet = preparedStatement.getGeneratedKeys()) { 3152 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 3153 startTime = nanoTime(); 3154 3155 while (resultSet.next()) { 3156 try { 3157 T listElement = resultSetMapperFor(statementContext).map(statementContext, resultSet, statementContext.getResultSetRowType().get(), getInstanceProvider()).orElse(null); 3158 list.add(listElement); 3159 } catch (SQLException e) { 3160 throw databaseExceptionWithStatementContext(statementContext, 3161 format("Unable to map JDBC generated-key row to %s", statementContext.getResultSetRowType().get()), e); 3162 } 3163 } 3164 3165 Duration resultSetMappingDuration = Duration.ofNanos(nanoTime() - startTime); 3166 StatementResult statementResult = getMetricsCollectorDispatcher().isEnabled() 3167 ? new StatementResult((long) list.size(), rowsAffected) 3168 : StatementResult.empty(); 3169 return new DatabaseOperationResult(executionDuration, resultSetMappingDuration, statementResult); 3170 } 3171 }, generatedKeysPreparedStatementFactory(requestedKeyColumnNames)); 3172 3173 return list; 3174 } 3175 3176 @NonNull 3177 private String[] copyGeneratedKeyColumnNames(@Nullable String @Nullable [] keyColumnNames) { 3178 if (keyColumnNames == null || keyColumnNames.length == 0) 3179 return new String[0]; 3180 3181 String[] copy = Arrays.copyOf(keyColumnNames, keyColumnNames.length); 3182 3183 for (String keyColumnName : copy) 3184 requireNonNull(keyColumnName); 3185 3186 return copy; 3187 } 3188 3189 @NonNull 3190 private PreparedStatementFactory generatedKeysPreparedStatementFactory(@NonNull String @NonNull [] keyColumnNames) { 3191 requireNonNull(keyColumnNames); 3192 3193 return (connection, statementContext) -> getDatabaseDialect(connection) 3194 .prepareGeneratedKeysStatement(connection, statementContext, keyColumnNames); 3195 } 3196 3197 /** 3198 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}, 3199 * which returns 0 or 1 rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING}, 3200 * MariaDB {@code INSERT ... RETURNING}, 3201 * or SQL Server {@code OUTPUT}. 3202 * 3203 * @param sql the SQL query to execute 3204 * @param resultSetRowType the type to which the {@link ResultSet} row should be marshaled 3205 * @param parameters {@link PreparedStatement} parameters, if any 3206 * @param <T> the type to be returned 3207 * @return a single result (or no result) 3208 * @throws DatabaseException if > 1 row is returned 3209 */ 3210 @NonNull 3211 private <T> Optional<T> executeForObject(@NonNull String sql, 3212 @NonNull Class<T> resultSetRowType, 3213 Object @Nullable ... parameters) { 3214 requireNonNull(sql); 3215 requireNonNull(resultSetRowType); 3216 3217 return executeForObject(Statement.of(generateId(), sql), resultSetRowType, parameters); 3218 } 3219 3220 /** 3221 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}, 3222 * which returns 0 or 1 rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING}, 3223 * MariaDB {@code INSERT ... RETURNING}, 3224 * or SQL Server {@code OUTPUT}. 3225 * 3226 * @param statement the SQL statement to execute 3227 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 3228 * @param parameters {@link PreparedStatement} parameters, if any 3229 * @param <T> the type to be returned 3230 * @return a single result (or no result) 3231 * @throws DatabaseException if > 1 row is returned 3232 */ 3233 private <T> Optional<T> executeForObject(@NonNull Statement statement, 3234 @NonNull Class<T> resultSetRowType, 3235 Object @Nullable ... parameters) { 3236 requireNonNull(statement); 3237 requireNonNull(resultSetRowType); 3238 3239 return executeForObject(statement, resultSetRowType, null, null, parameters); 3240 } 3241 3242 private <T> Optional<T> executeForObject(@NonNull Statement statement, 3243 @NonNull Class<T> resultSetRowType, 3244 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3245 StatementContext.@Nullable SpiOverrides spiOverrides, 3246 Object @Nullable ... parameters) { 3247 requireNonNull(statement); 3248 requireNonNull(resultSetRowType); 3249 3250 // Ultimately we just delegate to queryForObject. 3251 // Having `executeForList` is to allow for users to explicitly express intent 3252 // and make static analysis of code easier (e.g. maybe you'd like to hook all of your "execute" statements for 3253 // logging, or delegation to a writable master as opposed to a read replica) 3254 return queryForObject(statement, resultSetRowType, preparedStatementCustomizer, spiOverrides, parameters); 3255 } 3256 3257 /** 3258 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}, 3259 * which returns any number of rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING}, 3260 * MariaDB {@code INSERT ... RETURNING}, 3261 * or SQL Server {@code OUTPUT}. 3262 * 3263 * @param sql the SQL to execute 3264 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 3265 * @param parameters {@link PreparedStatement} parameters, if any 3266 * @param <T> the type to be returned 3267 * @return a list of results 3268 */ 3269 @NonNull 3270 private <T> List<@Nullable T> executeForList(@NonNull String sql, 3271 @NonNull Class<T> resultSetRowType, 3272 Object @Nullable ... parameters) { 3273 requireNonNull(sql); 3274 requireNonNull(resultSetRowType); 3275 3276 return executeForList(Statement.of(generateId(), sql), resultSetRowType, parameters); 3277 } 3278 3279 /** 3280 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE}, 3281 * which returns any number of rows with database-native syntax such as PostgreSQL/SQLite {@code RETURNING}, 3282 * MariaDB {@code INSERT ... RETURNING}, 3283 * or SQL Server {@code OUTPUT}. 3284 * 3285 * @param statement the SQL statement to execute 3286 * @param resultSetRowType the type to which {@link ResultSet} rows should be marshaled 3287 * @param parameters {@link PreparedStatement} parameters, if any 3288 * @param <T> the type to be returned 3289 * @return a list of results 3290 */ 3291 @NonNull 3292 private <T> List<@Nullable T> executeForList(@NonNull Statement statement, 3293 @NonNull Class<T> resultSetRowType, 3294 Object @Nullable ... parameters) { 3295 requireNonNull(statement); 3296 requireNonNull(resultSetRowType); 3297 3298 return executeForList(statement, resultSetRowType, null, null, parameters); 3299 } 3300 3301 private <T> List<@Nullable T> executeForList(@NonNull Statement statement, 3302 @NonNull Class<T> resultSetRowType, 3303 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3304 StatementContext.@Nullable SpiOverrides spiOverrides, 3305 Object @Nullable ... parameters) { 3306 requireNonNull(statement); 3307 requireNonNull(resultSetRowType); 3308 3309 // Ultimately we just delegate to queryForList. 3310 // Having `executeForList` is to allow for users to explicitly express intent 3311 // and make static analysis of code easier (e.g. maybe you'd like to hook all of your "execute" statements for 3312 // logging, or delegation to a writable master as opposed to a read replica) 3313 return queryForList(statement, resultSetRowType, preparedStatementCustomizer, spiOverrides, parameters); 3314 } 3315 3316 /** 3317 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE} 3318 * in "batch" over a set of parameter groups. 3319 * <p> 3320 * Useful for bulk-inserting or updating large amounts of data. 3321 * 3322 * @param sql the SQL to execute 3323 * @param parameterGroups Groups of {@link PreparedStatement} parameters 3324 * @return the number of rows affected by the SQL statement per-group 3325 */ 3326 @NonNull 3327 private List<Long> executeBatch(@NonNull String sql, 3328 @NonNull List<List<Object>> parameterGroups) { 3329 requireNonNull(sql); 3330 requireNonNull(parameterGroups); 3331 3332 return executeBatch(Statement.of(generateId(), sql), parameterGroups); 3333 } 3334 3335 /** 3336 * Executes a SQL Data Manipulation Language (DML) statement, such as {@code INSERT}, {@code UPDATE}, or {@code DELETE} 3337 * in "batch" over a set of parameter groups. 3338 * <p> 3339 * Useful for bulk-inserting or updating large amounts of data. 3340 * 3341 * @param statement the SQL statement to execute 3342 * @param parameterGroups Groups of {@link PreparedStatement} parameters 3343 * @return the number of rows affected by the SQL statement per-group 3344 */ 3345 @NonNull 3346 private List<Long> executeBatch(@NonNull Statement statement, 3347 @NonNull List<List<Object>> parameterGroups) { 3348 requireNonNull(statement); 3349 requireNonNull(parameterGroups); 3350 3351 return executeBatch(statement, parameterGroups, null, null, null); 3352 } 3353 3354 private List<Long> executeBatch(@NonNull Statement statement, 3355 @NonNull List<List<Object>> parameterGroups, 3356 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 3357 @Nullable Integer batchChunkSize, 3358 StatementContext.@Nullable SpiOverrides spiOverrides) { 3359 requireNonNull(statement); 3360 requireNonNull(parameterGroups); 3361 if (parameterGroups.isEmpty()) 3362 return List.of(); 3363 3364 Integer expectedParameterCount = null; 3365 3366 for (int i = 0; i < parameterGroups.size(); i++) { 3367 List<Object> parameterGroup = parameterGroups.get(i); 3368 3369 if (parameterGroup == null) 3370 throw new IllegalArgumentException(format("Parameter group at index %s is null", i)); 3371 3372 int parameterCount = parameterGroup.size(); 3373 if (expectedParameterCount == null) { 3374 expectedParameterCount = parameterCount; 3375 } else if (parameterCount != expectedParameterCount) { 3376 throw new IllegalArgumentException(format( 3377 "Inconsistent parameter group size at index %s: expected %s but found %s", 3378 i, expectedParameterCount, parameterCount)); 3379 } 3380 } 3381 3382 ResultHolder<List<Long>> resultHolder = new ResultHolder<>(); 3383 StatementContext<List<Long>> statementContext = StatementContext.with(statement, this) 3384 .parameters((List) parameterGroups) 3385 .resultSetRowType(List.class) 3386 .batchParameterGroups(true) 3387 .spiOverrides(spiOverrides) 3388 .build(); 3389 3390 if (batchChunkSize == null || batchChunkSize >= parameterGroups.size()) { 3391 performDatabaseOperation(statementContext, (preparedStatement) -> { 3392 applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer); 3393 3394 for (List<Object> parameterGroup : parameterGroups) { 3395 if (parameterGroup.size() > 0) 3396 performPreparedStatementBinding(statementContext, preparedStatement, parameterGroup); 3397 3398 preparedStatement.addBatch(); 3399 } 3400 }, (PreparedStatement preparedStatement) -> { 3401 long startTime = nanoTime(); 3402 List<Long> result = executePreparedStatementBatch(preparedStatement); 3403 3404 resultHolder.value = result; 3405 return batchDatabaseOperationResult(startTime, result); 3406 }, parameterGroups.size()); 3407 } else { 3408 int effectiveBatchChunkSize = batchChunkSize; 3409 3410 performDatabaseOperation(statementContext, (preparedStatement) -> { 3411 applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer); 3412 }, (PreparedStatement preparedStatement) -> { 3413 long startTime = nanoTime(); 3414 int currentBatchSize = 0; 3415 List<Long> result = new ArrayList<>(parameterGroups.size()); 3416 3417 for (List<Object> parameterGroup : parameterGroups) { 3418 if (parameterGroup.size() > 0) 3419 performPreparedStatementBinding(statementContext, preparedStatement, parameterGroup); 3420 3421 preparedStatement.addBatch(); 3422 ++currentBatchSize; 3423 3424 if (currentBatchSize == effectiveBatchChunkSize) { 3425 result.addAll(executePreparedStatementBatch(preparedStatement)); 3426 preparedStatement.clearBatch(); 3427 currentBatchSize = 0; 3428 } 3429 } 3430 3431 if (currentBatchSize > 0) { 3432 result.addAll(executePreparedStatementBatch(preparedStatement)); 3433 preparedStatement.clearBatch(); 3434 } 3435 3436 resultHolder.value = result; 3437 return batchDatabaseOperationResult(startTime, result); 3438 }, parameterGroups.size()); 3439 } 3440 3441 return resultHolder.value; 3442 } 3443 3444 @NonNull 3445 private DatabaseOperationResult batchDatabaseOperationResult(long startTime, 3446 @NonNull List<Long> result) { 3447 requireNonNull(result); 3448 3449 Duration executionDuration = Duration.ofNanos(nanoTime() - startTime); 3450 StatementResult statementResult = StatementResult.empty(); 3451 if (getMetricsCollectorDispatcher().isEnabled()) { 3452 Long rowsAffected = sumBatchUpdateCounts(result); 3453 statementResult = rowsAffected == null ? StatementResult.empty() : StatementResult.ofRowsAffected(rowsAffected); 3454 } 3455 3456 return new DatabaseOperationResult(executionDuration, null, statementResult); 3457 } 3458 3459 @NonNull 3460 private List<Long> executePreparedStatementBatch(@NonNull PreparedStatement preparedStatement) throws SQLException { 3461 requireNonNull(preparedStatement); 3462 3463 DatabaseOperationSupportStatus executeLargeBatchSupported = getExecuteLargeBatchSupported(); 3464 3465 // Use the appropriate "large" value if we know it. 3466 // If we don't know it, detect it and store it. 3467 if (executeLargeBatchSupported == DatabaseOperationSupportStatus.YES) { 3468 long[] resultArray = preparedStatement.executeLargeBatch(); 3469 return Arrays.stream(resultArray).boxed().collect(Collectors.toList()); 3470 } 3471 if (executeLargeBatchSupported == DatabaseOperationSupportStatus.NO) { 3472 int[] resultArray = preparedStatement.executeBatch(); 3473 return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList()); 3474 } 3475 3476 // If the driver doesn't support executeLargeBatch, then UnsupportedOperationException is thrown. 3477 try { 3478 long[] resultArray = preparedStatement.executeLargeBatch(); 3479 setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.YES); 3480 return Arrays.stream(resultArray).boxed().collect(Collectors.toList()); 3481 } catch (SQLFeatureNotSupportedException | UnsupportedOperationException | AbstractMethodError e) { 3482 setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.NO); 3483 int[] resultArray = preparedStatement.executeBatch(); 3484 return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList()); 3485 } catch (SQLException e) { 3486 if (!isUnsupportedSqlFeature(e)) 3487 throw e; 3488 3489 setExecuteLargeBatchSupported(DatabaseOperationSupportStatus.NO); 3490 int[] resultArray = preparedStatement.executeBatch(); 3491 return Arrays.stream(resultArray).asLongStream().boxed().collect(Collectors.toList()); 3492 } 3493 } 3494 3495 /** 3496 * Exposes a temporary handle to JDBC {@link DatabaseMetaData}, which provides comprehensive vendor-specific information about this database as a whole. 3497 * <p> 3498 * This method acquires {@link DatabaseMetaData} on its own newly-borrowed connection, which it manages internally. 3499 * <p> 3500 * It does <strong>not</strong> participate in the active transaction, if one exists. 3501 * <p> 3502 * The connection is closed as soon as {@link DatabaseMetaDataReader#read(DatabaseMetaData)} completes. 3503 * <p> 3504 * See <a href="https://docs.oracle.com/en/java/javase/26/docs/api/java.sql/java/sql/DatabaseMetaData.html">{@code DatabaseMetaData} Javadoc</a> for details. 3505 */ 3506 public void readDatabaseMetaData(@NonNull DatabaseMetaDataReader databaseMetaDataReader) { 3507 requireNonNull(databaseMetaDataReader); 3508 3509 performRawConnectionOperation((connection -> { 3510 databaseMetaDataReader.read(connection.getMetaData()); 3511 return Optional.empty(); 3512 }), false); 3513 } 3514 3515 /** 3516 * Performs raw JDBC work with a Pyranid-managed {@link Connection}. 3517 * <p> 3518 * If called inside a Pyranid transaction, this operation uses the transaction's connection and participates in that 3519 * transaction. Otherwise, Pyranid borrows a connection for the duration of the callback and closes it afterwards. 3520 * <p> 3521 * The {@link Connection} passed to {@code rawConnectionOperation} is a guarded handle. Normal JDBC operations are 3522 * delegated to the underlying driver connection, but lifecycle, transaction-management, and connection-wide state 3523 * methods such as 3524 * {@link Connection#close()}, {@link Connection#commit()}, {@link Connection#rollback()}, 3525 * {@link Connection#setAutoCommit(boolean)}, {@link Connection#setCatalog(String)}, {@link Connection#setSchema(String)}, and 3526 * {@link Connection#setNetworkTimeout(java.util.concurrent.Executor, int)} throw {@link IllegalStateException}. Use 3527 * Pyranid transaction APIs instead. 3528 * {@link Connection#unwrap(Class)} may return a guarded, callback-scoped proxy for a vendor interface, but never a 3529 * castable physical {@link Connection}; the proxy blocks lifecycle methods and expires with the callback. 3530 * JDBC objects created from this handle are also guarded: {@link java.sql.Statement#getConnection()} and 3531 * {@link java.sql.DatabaseMetaData#getConnection()} return the Pyranid-managed handle, and 3532 * {@link ResultSet#getStatement()} returns a guarded statement. Guarded statements, resultsets, and metadata refuse 3533 * driver-specific {@code unwrap(...)} calls that could expose the driver's underlying connection. 3534 * <p> 3535 * The connection handle is valid only for the duration of the callback. Do not close it, retain it, or use it after this 3536 * method returns. 3537 * 3538 * @param rawConnectionOperation the raw JDBC operation to perform 3539 * @param <T> the type to be returned 3540 * @return the operation result 3541 * @throws DatabaseException if connection acquisition, callback execution, or cleanup fails 3542 * @since 4.2.0 3543 */ 3544 @NonNull 3545 public <T> Optional<T> useRawConnection(@NonNull RawConnectionOperation<T> rawConnectionOperation) { 3546 requireNonNull(rawConnectionOperation); 3547 3548 return performRawConnectionOperation(connection -> { 3549 PyranidRawConnection rawConnection = new PyranidRawConnection(connection); 3550 3551 try { 3552 Optional<T> result = rawConnectionOperation.perform(rawConnection); 3553 return result == null ? Optional.empty() : result; 3554 } finally { 3555 rawConnection.release(); 3556 } 3557 }, true); 3558 } 3559 3560 /** 3561 * Performs an operation with one callback-scoped database-notification listener session. 3562 * <p> 3563 * This method is synchronous and blocking. It acquires at most one listener connection from this 3564 * {@code Database}'s configured {@link DataSource}, registers every requested channel, invokes {@code operation} 3565 * at most once, expires the supplied {@link NotificationSession}, and completes cleanup before returning or 3566 * throwing. It never reconnects. 3567 * <p> 3568 * The configured source must preserve one physical backend session for the entire checkout. For PostgreSQL, 3569 * direct connections and session pooling are suitable. Using PgBouncer transaction or statement pooling as the 3570 * listener source is unsupported: registration can appear to succeed before backend-session affinity is lost and 3571 * notification delivery silently stops. Pyranid does not inspect or validate proxy topology. Applications whose 3572 * ordinary source cannot provide the required affinity should construct a separate {@code Database} over a suitable 3573 * listener source and invoke this method on that instance. 3574 * <p> 3575 * Notifications are lossy hints. Durable applications should normally reconcile authoritative state as the first 3576 * callback action. The operation may use ordinary database methods, but those methods acquire or select their 3577 * connection normally and never reuse the listener connection. Pyranid transaction entry is interrupt-sensitive; 3578 * when a receive returns a batch with a racing interrupt, temporarily clear and remember the flag during bounded 3579 * reconciliation and restore it afterward as described by {@link NotificationSession#awaitNotifications(Duration)}. 3580 * <p> 3581 * A terminal receive failure is retained by the session and rethrown after cleanup even if {@code operation} catches 3582 * it and returns. A retained transport {@link Error} propagates as that exact, unwrapped instance. If the callback 3583 * instead throws a distinct {@code Error}, that callback error remains primary and the retained transport failure is 3584 * suppressed beneath it. 3585 * 3586 * @param channels fixed, nonempty set of nonblank channels to register 3587 * @param operation operation to invoke after every channel has been registered 3588 * @throws NullPointerException if {@code channels}, a channel, or {@code operation} is null 3589 * @throws IllegalArgumentException if the set is empty or a channel violates common or backend-specific limits 3590 * @throws IllegalStateException if any Pyranid transaction is active on the calling thread 3591 * @throws InterruptedException if cooperative interruption wins after any required cleanup 3592 * @throws UnsupportedOperationException if the database dialect or runtime driver cannot receive notifications 3593 * @throws DatabaseException if connection acquisition, setup, receive, callback execution, or cleanup fails 3594 * @since 4.6.0 3595 */ 3596 public void withNotificationSession(@NonNull Set<@NonNull String> channels, 3597 @NonNull NotificationSessionOperation operation) 3598 throws InterruptedException { 3599 requireNonNull(channels); 3600 requireNonNull(operation); 3601 3602 if (channels.isEmpty()) 3603 throw new IllegalArgumentException("channels must not be empty"); 3604 3605 Set<String> validatedChannels = new LinkedHashSet<>(channels.size()); 3606 3607 for (String channel : channels) { 3608 Notification.validateChannel(channel); 3609 validatedChannels.add(channel); 3610 } 3611 3612 if (hasAmbientTransaction()) 3613 throw new IllegalStateException("Notification sessions are not permitted inside a Pyranid transaction"); 3614 3615 if (Thread.interrupted()) 3616 throw new InterruptedException(); 3617 3618 DatabaseType databaseType = getDatabaseType(); 3619 DatabaseNotificationSupport notificationSupport = getDatabaseDialect().notificationSupport(); 3620 3621 for (String channel : validatedChannels) 3622 notificationSupport.validateChannel(channel); 3623 3624 if (Thread.interrupted()) 3625 throw new InterruptedException(); 3626 3627 performNotificationSession(Set.copyOf(validatedChannels), operation, databaseType, notificationSupport); 3628 } 3629 3630 /** 3631 * Performs an operation with one callback-scoped database-notification listener session for a single channel. 3632 * <p> 3633 * This is the single-channel convenience form of 3634 * {@link #withNotificationSession(Set, NotificationSessionOperation)}. The listener connection comes from this 3635 * {@code Database}'s configured {@link DataSource}, which must preserve physical backend-session affinity for the 3636 * entire checkout. For PostgreSQL, using PgBouncer transaction or statement pooling as the listener source is 3637 * unsupported: registration can appear to succeed before affinity is lost and delivery silently stops. Pyranid does 3638 * not inspect or validate proxy topology. 3639 * <p> 3640 * A terminal receive failure is retained and rethrown after cleanup even if {@code operation} catches it and returns. 3641 * A retained transport {@link Error} propagates as the exact, unwrapped instance unless a distinct callback 3642 * {@code Error} takes precedence as described by the set-based overload. 3643 * 3644 * @param channel nonblank channel to register 3645 * @param operation operation to invoke after the channel has been registered 3646 * @throws NullPointerException if {@code channel} or {@code operation} is null 3647 * @throws IllegalArgumentException if the channel violates common or backend-specific limits 3648 * @throws IllegalStateException if any Pyranid transaction is active on the calling thread 3649 * @throws InterruptedException if cooperative interruption wins after any required cleanup 3650 * @throws UnsupportedOperationException if the database dialect or runtime driver cannot receive notifications 3651 * @throws DatabaseException if connection acquisition, setup, receive, callback execution, or cleanup fails 3652 * @since 4.6.0 3653 */ 3654 public void withNotificationSession(@NonNull String channel, 3655 @NonNull NotificationSessionOperation operation) 3656 throws InterruptedException { 3657 requireNonNull(channel); 3658 requireNonNull(operation); 3659 withNotificationSession(Set.of(channel), operation); 3660 } 3661 3662 /** 3663 * Sends a transient database notification without specifying a payload. 3664 * <p> 3665 * Sending follows ordinary Pyranid statement and transaction selection. On PostgreSQL, a send inside a Pyranid 3666 * transaction becomes visible only if that transaction commits. Payload representation is database-specific; 3667 * PostgreSQL converts the resulting null payload to the empty string. 3668 * 3669 * @param channel nonblank notification channel 3670 * @throws NullPointerException if {@code channel} is null 3671 * @throws IllegalArgumentException if the channel violates common or backend-specific limits 3672 * @throws UnsupportedOperationException if the database dialect does not support notification sends 3673 * @throws DatabaseException if the send fails 3674 * @since 4.6.0 3675 */ 3676 public void sendNotification(@NonNull String channel) { 3677 sendNotification(channel, null); 3678 } 3679 3680 /** 3681 * Sends a transient database notification. 3682 * <p> 3683 * Sending follows ordinary Pyranid statement and transaction selection, including connection ownership, 3684 * parameter binding and redaction, statement logging, timeout configuration, and metrics. Payload nullability 3685 * and null/empty-string handling are database-specific; Pyranid performs no generic normalization. 3686 * <p> 3687 * On PostgreSQL this executes bound {@code pg_notify(?, ?)} SQL. PostgreSQL converts a null payload to the empty 3688 * string. Delivery occurs only after commit and is discarded by rollback; notification delivery itself remains 3689 * non-durable and may be coalesced. 3690 * 3691 * @param channel nonblank notification channel 3692 * @param payload notification payload, possibly null or empty 3693 * @throws NullPointerException if {@code channel} is null 3694 * @throws IllegalArgumentException if the channel or payload violates common or backend-specific limits 3695 * @throws UnsupportedOperationException if the database dialect does not support notification sends 3696 * @throws DatabaseException if the send fails 3697 * @since 4.6.0 3698 */ 3699 public void sendNotification(@NonNull String channel, 3700 @Nullable String payload) { 3701 Notification.validateChannel(channel); 3702 DatabaseNotificationSupport notificationSupport = getDatabaseDialect().notificationSupport(); 3703 3704 notificationSupport.validateChannel(channel); 3705 notificationSupport.validatePayload(payload); 3706 3707 if (!notificationSupport.isSendSupported()) 3708 throw new UnsupportedOperationException(format( 3709 "Database type %s does not support notification sends", getDatabaseType())); 3710 3711 Statement statement = Statement.of(generateId(), notificationSupport.sendStatementSql()); 3712 StatementContext<Void> statementContext = StatementContext.with(statement, this) 3713 .parameters(channel, payload) 3714 .build(); 3715 3716 PreparedStatementCustomizer defaultPreparedStatementCustomizer = hasDefaultPreparedStatementSettings() 3717 ? (context, preparedStatement) -> applyDefaultPreparedStatementSettings(preparedStatement) 3718 : null; 3719 3720 performDatabaseOperation(statementContext, Arrays.asList(channel, payload), 3721 defaultPreparedStatementCustomizer, preparedStatement -> { 3722 long startTime = nanoTime(); 3723 executeAndDrain(preparedStatement); 3724 return new DatabaseOperationResult( 3725 Duration.ofNanos(nanoTime() - startTime), null, StatementResult.empty()); 3726 }); 3727 } 3728 3729 /** 3730 * Reports whether the configured database dialect and currently loadable runtime adapter expose the APIs required 3731 * to attempt a notification-listening session. 3732 * <p> 3733 * This method resolves the full database type. If it has not been configured or cached, resolution may acquire a 3734 * metadata connection and may throw {@link DatabaseException}. It does not acquire a listener session, inspect pool 3735 * or proxy mode, prove backend-session affinity, unwrap a physical listener connection, or emit notification-session 3736 * lifecycle metrics. 3737 * 3738 * @return true if notification listening can be attempted with the current dialect and runtime 3739 * @throws DatabaseException if automatic database-type detection fails 3740 * @since 4.6.0 3741 */ 3742 @NonNull 3743 public Boolean isNotificationListeningSupported() { 3744 return getDatabaseDialect().notificationSupport().isReceiveRuntimeAvailable(); 3745 } 3746 3747 private void executeAndDrain(@NonNull PreparedStatement preparedStatement) throws SQLException { 3748 requireNonNull(preparedStatement); 3749 3750 boolean resultAvailable = preparedStatement.execute(); 3751 3752 for (;;) { 3753 if (resultAvailable) { 3754 try (ResultSet resultSet = preparedStatement.getResultSet()) { 3755 if (resultSet != null) { 3756 while (resultSet.next()) { 3757 // Drain every row before advancing to a possible subsequent result. 3758 } 3759 } 3760 } 3761 } else if (preparedStatement.getUpdateCount() == -1) { 3762 break; 3763 } 3764 3765 resultAvailable = preparedStatement.getMoreResults(java.sql.Statement.CLOSE_CURRENT_RESULT); 3766 } 3767 } 3768 3769 private void performNotificationSession(@NonNull Set<@NonNull String> channels, 3770 @NonNull NotificationSessionOperation operation, 3771 @NonNull DatabaseType databaseType, 3772 @NonNull DatabaseNotificationSupport notificationSupport) 3773 throws InterruptedException { 3774 requireNonNull(channels); 3775 requireNonNull(operation); 3776 requireNonNull(databaseType); 3777 requireNonNull(notificationSupport); 3778 3779 MetricsCollectorDispatcher metricsCollectorDispatcher = getMetricsCollectorDispatcher(); 3780 UUID notificationSessionId = metricsCollectorDispatcher.isEnabled() ? UUID.randomUUID() : null; 3781 long lifecycleStartTime = notificationSessionId == null ? 0L : nanoTime(); 3782 3783 if (notificationSessionId != null) 3784 metricsCollectorDispatcher.willOpenNotificationSession(databaseType, notificationSessionId); 3785 3786 Connection connection = null; 3787 NotificationTransport transport = null; 3788 NotificationSession session = null; 3789 Boolean initialAutoCommit = null; 3790 NotificationSetupPhase setupPhase = NotificationSetupPhase.CAPABILITY; 3791 UnsupportedOperationException frameworkUnsupportedFailure = null; 3792 boolean initialSanitationComplete = false; 3793 3794 try { 3795 if (!notificationSupport.isReceiveRuntimeAvailable()) { 3796 frameworkUnsupportedFailure = new UnsupportedOperationException(format( 3797 "Database type %s does not support notification listening with the current runtime", databaseType)); 3798 throw frameworkUnsupportedFailure; 3799 } 3800 3801 setupPhase = NotificationSetupPhase.ACQUIRE_CONNECTION; 3802 connection = requireNonNull(getDataSource().getConnection(), "DataSource returned a null connection"); 3803 3804 setupPhase = NotificationSetupPhase.CONFIGURE_CONNECTION; 3805 initialAutoCommit = connection.getAutoCommit(); 3806 3807 if (!initialAutoCommit) { 3808 // A pooled connection can be returned with an unfinished transaction. Discard inherited work before 3809 // enabling autocommit; LISTEN registration must not inherit that transaction. 3810 connection.rollback(); 3811 connection.setAutoCommit(true); 3812 } 3813 3814 setupPhase = NotificationSetupPhase.OPEN_TRANSPORT; 3815 transport = notificationSupport.open(connection); 3816 3817 setupPhase = NotificationSetupPhase.SANITIZE_UNLISTEN; 3818 transport.unlistenAll(); 3819 3820 setupPhase = NotificationSetupPhase.SANITIZE_DRAIN; 3821 requireNonNull(transport.drain(), "Notification transport returned a null batch"); 3822 3823 if (transport.isConnectionUncertain()) 3824 throw new SQLException("Notification transport became uncertain during initial sanitation"); 3825 3826 initialSanitationComplete = true; 3827 setupPhase = NotificationSetupPhase.REGISTER; 3828 transport.listen(channels); 3829 3830 if (transport.isConnectionUncertain()) 3831 throw new SQLException("Notification transport became uncertain during registration"); 3832 3833 session = new NotificationSession( 3834 this, requireNonNull(transport), databaseType, notificationSessionId); 3835 } catch (Throwable setupCause) { 3836 boolean connectionFailure = isNotificationConnectionFailure(connection, setupCause); 3837 Throwable setupFailure = normalizeNotificationSetupFailure( 3838 setupCause, frameworkUnsupportedFailure, setupPhase, connectionFailure, databaseType); 3839 3840 if (notificationSessionId != null) { 3841 metricsCollectorDispatcher.didFailToOpenNotificationSession( 3842 databaseType, notificationSessionId, 3843 Duration.ofNanos(nanoTime() - lifecycleStartTime), setupFailure); 3844 } 3845 3846 boolean interruptionObserved = Thread.interrupted(); 3847 // Synthetic uncertainty failures need not carry a connection-class SQLSTATE. Cleanup still fails closed: 3848 // this independent transport-state check prevents reuse even if generic exception classification does not. 3849 boolean canAttemptHealthyCleanup = initialSanitationComplete 3850 && connection != null 3851 && transport != null 3852 && initialAutoCommit != null 3853 && setupPhase == NotificationSetupPhase.REGISTER 3854 && !connectionFailure 3855 && !transport.isConnectionUncertain(); 3856 Throwable cleanupFailure = canAttemptHealthyCleanup 3857 ? cleanupOpenedNotificationSession(connection, transport, initialAutoCommit, false, databaseType) 3858 : cleanupFailedNotificationCandidate(connection, databaseType); 3859 setupFailure = appendSuppressedIfPresent(setupFailure, cleanupFailure); 3860 3861 if (Thread.interrupted()) 3862 interruptionObserved = true; 3863 3864 if (interruptionObserved) 3865 Thread.currentThread().interrupt(); 3866 3867 throwNotificationFailure(setupFailure); 3868 return; 3869 } 3870 3871 if (notificationSessionId != null) { 3872 metricsCollectorDispatcher.didOpenNotificationSession( 3873 databaseType, notificationSessionId, 3874 Duration.ofNanos(nanoTime() - lifecycleStartTime)); 3875 } 3876 3877 session = requireNonNull(session); 3878 Throwable callbackFailure = null; 3879 boolean callbackFailureIsError = false; 3880 InterruptedException concreteInterruptedException = null; 3881 boolean interruptionObserved = false; 3882 3883 try { 3884 if (Thread.interrupted()) { 3885 interruptionObserved = true; 3886 } else { 3887 operation.perform(session); 3888 } 3889 } catch (InterruptedException interruptedException) { 3890 concreteInterruptedException = interruptedException; 3891 interruptionObserved = true; 3892 // Retain the exact exception while clearing any simultaneously pending status before cleanup. 3893 Thread.interrupted(); 3894 } catch (DatabaseException databaseException) { 3895 callbackFailure = databaseException; 3896 } catch (Error error) { 3897 callbackFailure = error; 3898 callbackFailureIsError = true; 3899 } catch (Throwable throwable) { 3900 callbackFailure = new DatabaseException( 3901 "Notification session operation failed", throwable, databaseType.dialect()); 3902 } finally { 3903 session.expire(); 3904 } 3905 3906 Throwable transportFailure = session.terminalFailure(); 3907 Throwable primaryFailure; 3908 3909 if (callbackFailureIsError) { 3910 primaryFailure = appendSuppressedIfPresent(callbackFailure, transportFailure); 3911 } else if (transportFailure != null) { 3912 primaryFailure = appendSuppressedIfPresent(transportFailure, callbackFailure); 3913 } else { 3914 primaryFailure = callbackFailure; 3915 } 3916 3917 // Cleanup begins flag-clear with respect to status already pending at this boundary. 3918 if (Thread.interrupted()) 3919 interruptionObserved = true; 3920 3921 Throwable cleanupFailure = cleanupOpenedNotificationSession( 3922 requireNonNull(connection), requireNonNull(transport), requireNonNull(initialAutoCommit), 3923 session.isConnectionUncertain(), databaseType); 3924 primaryFailure = appendSuppressedIfPresent(primaryFailure, cleanupFailure); 3925 3926 if (Thread.interrupted()) 3927 interruptionObserved = true; 3928 3929 MetricsCollector.NotificationSessionOutcome outcome; 3930 3931 if (primaryFailure != null) 3932 outcome = MetricsCollector.NotificationSessionOutcome.FAILED; 3933 else if (interruptionObserved) 3934 outcome = MetricsCollector.NotificationSessionOutcome.INTERRUPTED; 3935 else 3936 outcome = MetricsCollector.NotificationSessionOutcome.CALLBACK_RETURNED; 3937 3938 if (notificationSessionId != null) { 3939 metricsCollectorDispatcher.didCloseNotificationSession( 3940 databaseType, notificationSessionId, outcome, 3941 Duration.ofNanos(nanoTime() - lifecycleStartTime), 3942 outcome == MetricsCollector.NotificationSessionOutcome.FAILED ? primaryFailure : null); 3943 } 3944 3945 if (primaryFailure != null) { 3946 if (interruptionObserved) { 3947 primaryFailure = appendSuppressedIfPresent(primaryFailure, concreteInterruptedException); 3948 Thread.currentThread().interrupt(); 3949 } 3950 3951 throwNotificationFailure(primaryFailure); 3952 return; 3953 } 3954 3955 if (interruptionObserved) 3956 throw concreteInterruptedException == null ? new InterruptedException() : concreteInterruptedException; 3957 } 3958 3959 @NonNull 3960 private Throwable normalizeNotificationSetupFailure(@NonNull Throwable setupCause, 3961 @Nullable UnsupportedOperationException frameworkUnsupportedFailure, 3962 @NonNull NotificationSetupPhase setupPhase, 3963 boolean connectionFailure, 3964 @NonNull DatabaseType databaseType) { 3965 requireNonNull(setupCause); 3966 requireNonNull(setupPhase); 3967 requireNonNull(databaseType); 3968 3969 if (setupCause == frameworkUnsupportedFailure) 3970 return setupCause; 3971 3972 if (setupCause instanceof NotificationReceiveUnsupportedException && !connectionFailure) { 3973 return new UnsupportedOperationException( 3974 "Notification listening is not supported by the current database driver connection", setupCause); 3975 } 3976 3977 if (setupPhase == NotificationSetupPhase.OPEN_TRANSPORT 3978 && !connectionFailure 3979 && (setupCause instanceof SQLFeatureNotSupportedException 3980 || setupCause instanceof UnsupportedOperationException 3981 || setupCause instanceof AbstractMethodError)) { 3982 return new UnsupportedOperationException( 3983 "Notification listening is not supported by the current database driver connection", setupCause); 3984 } 3985 3986 return normalizeNotificationFailure("Unable to open notification session", setupCause, databaseType); 3987 } 3988 3989 private boolean isNotificationConnectionFailure(@Nullable Connection connection, 3990 @NonNull Throwable throwable) { 3991 requireNonNull(throwable); 3992 3993 if (throwable instanceof SQLException sqlException) { 3994 String sqlState = sqlException.getSQLState(); 3995 3996 if (sqlState != null && sqlState.startsWith("08")) 3997 return true; 3998 } 3999 4000 if (connection != null) { 4001 try { 4002 return connection.isClosed(); 4003 } catch (Throwable diagnosticFailure) { 4004 if (diagnosticFailure != throwable) 4005 throwable.addSuppressed(diagnosticFailure); 4006 4007 if (diagnosticFailure instanceof SQLException diagnosticSqlException) { 4008 String sqlState = diagnosticSqlException.getSQLState(); 4009 4010 if (sqlState != null && sqlState.startsWith("08")) 4011 return true; 4012 } 4013 4014 // If connection health cannot be inspected, never return the candidate to a pool as healthy. 4015 return true; 4016 } 4017 } 4018 4019 return false; 4020 } 4021 4022 @Nullable 4023 private Throwable cleanupFailedNotificationCandidate(@Nullable Connection connection, 4024 @NonNull DatabaseType databaseType) { 4025 requireNonNull(databaseType); 4026 4027 if (connection == null) 4028 return null; 4029 4030 Throwable cleanupFailure = null; 4031 4032 try { 4033 connection.abort(Runnable::run); 4034 } catch (Throwable throwable) { 4035 cleanupFailure = appendSuppressed(cleanupFailure, 4036 normalizeNotificationFailure("Unable to abort failed notification-session connection", 4037 throwable, databaseType)); 4038 } 4039 4040 try { 4041 connection.close(); 4042 } catch (Throwable throwable) { 4043 cleanupFailure = appendSuppressed(cleanupFailure, 4044 normalizeNotificationFailure("Unable to close failed notification-session connection", 4045 throwable, databaseType)); 4046 } 4047 4048 return cleanupFailure; 4049 } 4050 4051 @Nullable 4052 private Throwable cleanupOpenedNotificationSession(@NonNull Connection connection, 4053 @NonNull NotificationTransport transport, 4054 @NonNull Boolean initialAutoCommit, 4055 boolean connectionUncertain, 4056 @NonNull DatabaseType databaseType) { 4057 requireNonNull(connection); 4058 requireNonNull(transport); 4059 requireNonNull(initialAutoCommit); 4060 requireNonNull(databaseType); 4061 4062 Throwable cleanupFailure = null; 4063 boolean abortRequired = connectionUncertain; 4064 4065 if (!abortRequired) { 4066 try { 4067 transport.unlistenAllForCleanup(); 4068 } catch (Throwable throwable) { 4069 cleanupFailure = appendSuppressed(cleanupFailure, 4070 normalizeNotificationFailure("Unable to unregister notification channels", throwable, databaseType)); 4071 abortRequired = true; 4072 } 4073 } 4074 4075 if (!abortRequired) { 4076 try { 4077 requireNonNull(transport.drain(), "Notification transport returned a null cleanup batch"); 4078 4079 if (transport.isConnectionUncertain()) 4080 throw new SQLException("Notification transport became uncertain during cleanup"); 4081 } catch (Throwable throwable) { 4082 cleanupFailure = appendSuppressed(cleanupFailure, 4083 normalizeNotificationFailure("Unable to drain notification connection during cleanup", 4084 throwable, databaseType)); 4085 abortRequired = true; 4086 } 4087 } 4088 4089 if (!abortRequired && !initialAutoCommit) { 4090 try { 4091 connection.setAutoCommit(false); 4092 } catch (Throwable throwable) { 4093 cleanupFailure = appendSuppressed(cleanupFailure, 4094 normalizeNotificationFailure("Unable to restore notification connection autocommit", 4095 throwable, databaseType)); 4096 abortRequired = true; 4097 } 4098 } 4099 4100 if (!abortRequired) { 4101 try { 4102 connection.close(); 4103 } catch (Throwable throwable) { 4104 cleanupFailure = appendSuppressed(cleanupFailure, 4105 normalizeNotificationFailure("Unable to close notification-session connection", 4106 throwable, databaseType)); 4107 // Once close has been invoked, the connection handle might already have been returned to a pool. 4108 // Do not attempt abort or a second close through a handle whose ownership is now unknown. 4109 } 4110 } 4111 4112 if (abortRequired) { 4113 try { 4114 connection.abort(Runnable::run); 4115 } catch (Throwable throwable) { 4116 cleanupFailure = appendSuppressed(cleanupFailure, 4117 normalizeNotificationFailure("Unable to abort uncertain notification-session connection", 4118 throwable, databaseType)); 4119 } 4120 4121 try { 4122 connection.close(); 4123 } catch (Throwable throwable) { 4124 cleanupFailure = appendSuppressed(cleanupFailure, 4125 normalizeNotificationFailure("Unable to close uncertain notification-session connection", 4126 throwable, databaseType)); 4127 } 4128 } 4129 4130 return cleanupFailure; 4131 } 4132 4133 @NonNull 4134 private Throwable normalizeNotificationFailure(@NonNull String message, 4135 @NonNull Throwable throwable, 4136 @NonNull DatabaseType databaseType) { 4137 requireNonNull(message); 4138 requireNonNull(throwable); 4139 requireNonNull(databaseType); 4140 4141 if (throwable instanceof DatabaseException || throwable instanceof Error) 4142 return throwable; 4143 4144 return new DatabaseException(message, throwable, databaseType.dialect()); 4145 } 4146 4147 private static void throwNotificationFailure(@NonNull Throwable throwable) { 4148 requireNonNull(throwable); 4149 4150 if (throwable instanceof RuntimeException runtimeException) 4151 throw runtimeException; 4152 4153 if (throwable instanceof Error error) 4154 throw error; 4155 4156 throw new DatabaseException("Notification session failed", throwable); 4157 } 4158 4159 @Nullable 4160 private static Throwable appendSuppressedIfPresent(@Nullable Throwable existing, 4161 @Nullable Throwable additional) { 4162 if (additional == null) 4163 return existing; 4164 4165 return appendSuppressed(existing, additional); 4166 } 4167 4168 private enum NotificationSetupPhase { 4169 CAPABILITY, 4170 ACQUIRE_CONNECTION, 4171 CONFIGURE_CONNECTION, 4172 OPEN_TRANSPORT, 4173 SANITIZE_UNLISTEN, 4174 SANITIZE_DRAIN, 4175 REGISTER 4176 } 4177 4178 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4179 @NonNull List<Object> parameters, 4180 @NonNull DatabaseOperation databaseOperation) { 4181 requireNonNull(statementContext); 4182 requireNonNull(parameters); 4183 requireNonNull(databaseOperation); 4184 4185 performDatabaseOperation(statementContext, parameters, null, databaseOperation); 4186 } 4187 4188 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4189 @NonNull List<Object> parameters, 4190 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 4191 @NonNull DatabaseOperation databaseOperation) { 4192 performDatabaseOperation(statementContext, parameters, preparedStatementCustomizer, databaseOperation, 4193 (connection, context) -> connection.prepareStatement(context.getStatement().getSql())); 4194 } 4195 4196 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4197 @NonNull List<Object> parameters, 4198 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 4199 @NonNull DatabaseOperation databaseOperation, 4200 @NonNull PreparedStatementFactory preparedStatementFactory) { 4201 requireNonNull(statementContext); 4202 requireNonNull(parameters); 4203 requireNonNull(databaseOperation); 4204 requireNonNull(preparedStatementFactory); 4205 4206 performDatabaseOperation(statementContext, (preparedStatement) -> { 4207 applyPreparedStatementCustomizer(statementContext, preparedStatement, preparedStatementCustomizer); 4208 if (parameters.size() > 0) 4209 performPreparedStatementBinding(statementContext, preparedStatement, parameters); 4210 }, databaseOperation, null, preparedStatementFactory); 4211 } 4212 4213 private <T> void performPreparedStatementBinding(@NonNull StatementContext<T> statementContext, 4214 @NonNull PreparedStatement preparedStatement, 4215 @NonNull List<Object> parameters) { 4216 requireNonNull(statementContext); 4217 requireNonNull(preparedStatement); 4218 requireNonNull(parameters); 4219 4220 try { 4221 DefaultPreparedStatementBinder.ParameterSqlTypeResolver parameterSqlTypeResolver = 4222 new DefaultPreparedStatementBinder.ParameterSqlTypeResolver(preparedStatement); 4223 PreparedStatementBinder preparedStatementBinder = preparedStatementBinderFor(statementContext); 4224 4225 for (int i = 0; i < parameters.size(); ++i) { 4226 Object parameter = SecureParameterSupport.unwrapSecureAndOptionalParameter(parameters.get(i)); 4227 Integer parameterIndex = i + 1; 4228 4229 if (parameter != null) { 4230 if (preparedStatementBinder instanceof DefaultPreparedStatementBinder defaultPreparedStatementBinder) { 4231 defaultPreparedStatementBinder.bindParameter(statementContext, preparedStatement, parameterIndex, parameter, 4232 parameterSqlTypeResolver); 4233 } else { 4234 preparedStatementBinder.bindParameter(statementContext, preparedStatement, parameterIndex, parameter); 4235 } 4236 } else { 4237 Integer sqlType = parameterSqlTypeResolver.determineParameterSqlType(parameterIndex) 4238 .map(DefaultPreparedStatementBinder.ParameterSqlType::getSqlType) 4239 .orElse(Types.NULL); 4240 try { 4241 preparedStatement.setNull(parameterIndex, sqlType); 4242 } catch (SQLException | AbstractMethodError e) { 4243 if (sqlType == Types.NULL) 4244 throw e; 4245 4246 preparedStatement.setNull(parameterIndex, Types.NULL); 4247 } 4248 } 4249 } 4250 } catch (Exception e) { 4251 throw databaseExceptionWithStatementContext(statementContext, e); 4252 } 4253 } 4254 4255 private void applyPreparedStatementCustomizer(@NonNull StatementContext<?> statementContext, 4256 @NonNull PreparedStatement preparedStatement, 4257 @Nullable PreparedStatementCustomizer preparedStatementCustomizer) throws SQLException { 4258 requireNonNull(statementContext); 4259 requireNonNull(preparedStatement); 4260 4261 if (preparedStatementCustomizer == null) 4262 return; 4263 4264 preparedStatementCustomizer.customize(statementContext, preparedStatement); 4265 } 4266 4267 private boolean hasDefaultPreparedStatementSettings() { 4268 return this.queryTimeout != null || this.fetchSize != null || this.maxRows != null; 4269 } 4270 4271 private void applyDefaultPreparedStatementSettings(@NonNull PreparedStatement preparedStatement) throws SQLException { 4272 requireNonNull(preparedStatement); 4273 applyPreparedStatementSettings(preparedStatement, this.queryTimeout, this.fetchSize, this.maxRows); 4274 } 4275 4276 private static void applyPreparedStatementSettings(@NonNull PreparedStatement preparedStatement, 4277 @Nullable Duration queryTimeout, 4278 @Nullable Integer fetchSize, 4279 @Nullable Integer maxRows) throws SQLException { 4280 requireNonNull(preparedStatement); 4281 4282 if (queryTimeout != null) 4283 preparedStatement.setQueryTimeout(queryTimeoutSeconds(queryTimeout)); 4284 4285 if (fetchSize != null) 4286 preparedStatement.setFetchSize(fetchSize); 4287 4288 if (maxRows != null) 4289 preparedStatement.setMaxRows(maxRows); 4290 } 4291 4292 @FunctionalInterface 4293 interface InternalRawConnectionOperation<R> { 4294 @NonNull 4295 Optional<R> perform(@NonNull Connection connection) throws Exception; 4296 } 4297 4298 /** 4299 * Gets the database type for this database. 4300 * <p> 4301 * If {@link Builder#databaseType(DatabaseType)} was not configured and the database type has not already been detected, 4302 * this method may acquire a connection and inspect {@link DatabaseMetaData}. Configure an explicit database type to avoid 4303 * runtime detection. 4304 * 4305 * @return the database type 4306 * @throws DatabaseException if automatic database type detection fails 4307 * @since 3.0.0 4308 */ 4309 @NonNull 4310 public DatabaseType getDatabaseType() { 4311 return getDatabaseType(this.databaseTypeDetectionConnectionHolder.get()); 4312 } 4313 4314 @NonNull 4315 DatabaseType peekDatabaseType() { 4316 DatabaseType cachedDatabaseType = this.databaseType.get(); 4317 return cachedDatabaseType == null ? DatabaseType.GENERIC : cachedDatabaseType; 4318 } 4319 4320 @NonNull 4321 DatabaseDialect getDatabaseDialect() { 4322 return getDatabaseDialect(this.databaseTypeDetectionConnectionHolder.get()); 4323 } 4324 4325 @NonNull 4326 DatabaseDialect getDatabaseDialect(@Nullable Connection connection) { 4327 DatabaseDialect cachedDatabaseDialect = this.databaseDialect.get(); 4328 4329 if (cachedDatabaseDialect != null) 4330 return cachedDatabaseDialect; 4331 4332 DatabaseDialect detectedDatabaseDialect = getDatabaseType(connection).dialect(); 4333 4334 if (this.databaseDialect.compareAndSet(null, detectedDatabaseDialect)) 4335 return detectedDatabaseDialect; 4336 4337 return requireNonNull(this.databaseDialect.get()); 4338 } 4339 4340 private void warmDatabaseTypeCacheForMetricsIfNeeded(@NonNull StatementContext<?> statementContext) { 4341 requireNonNull(statementContext); 4342 4343 if (!getMetricsCollectorDispatcher().isEnabled()) 4344 return; 4345 4346 // Trigger lazy database-type detection while the statement's JDBC connection is active so later 4347 // transaction-scope metrics can report an accurate db.system.name without opening a second metadata connection. 4348 try { 4349 statementContext.getDatabaseType(); 4350 } catch (Throwable t) { 4351 this.logger.log(Level.FINE, "Unable to warm database type cache for metrics", t); 4352 } 4353 } 4354 4355 private void dispatchWithDatabaseTypeDetectionConnection(@NonNull Connection connection, 4356 @NonNull Runnable operation) { 4357 requireNonNull(connection); 4358 requireNonNull(operation); 4359 4360 Connection previousDatabaseTypeDetectionConnection = this.databaseTypeDetectionConnectionHolder.get(); 4361 this.databaseTypeDetectionConnectionHolder.set(connection); 4362 4363 try { 4364 operation.run(); 4365 } finally { 4366 if (previousDatabaseTypeDetectionConnection == null) 4367 this.databaseTypeDetectionConnectionHolder.remove(); 4368 else 4369 this.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection); 4370 } 4371 } 4372 4373 @NonNull 4374 private DatabaseType getDatabaseType(@Nullable Connection connection) { 4375 DatabaseType cachedDatabaseType = this.databaseType.get(); 4376 4377 if (cachedDatabaseType != null) 4378 return cachedDatabaseType; 4379 4380 DatabaseType detectedDatabaseType; 4381 4382 try { 4383 detectedDatabaseType = connection == null 4384 ? DatabaseType.fromDataSource(getDataSource()) 4385 : DatabaseType.fromConnection(connection); 4386 } catch (DatabaseException e) { 4387 throw new DatabaseException(format( 4388 "Unable to determine database type automatically. Configure %s.%s(%s) explicitly to avoid runtime detection.", 4389 Builder.class.getSimpleName(), "databaseType", DatabaseType.class.getSimpleName()), e); 4390 } 4391 4392 if (this.databaseType.compareAndSet(null, detectedDatabaseType)) 4393 return detectedDatabaseType; 4394 4395 return this.databaseType.get(); 4396 } 4397 4398 /** 4399 * @since 3.0.0 4400 */ 4401 @NonNull 4402 public ZoneId getTimeZone() { 4403 return this.timeZone; 4404 } 4405 4406 /** 4407 * How should Pyranid bind {@link java.time.Instant} and {@link java.time.OffsetDateTime} parameters when JDBC 4408 * parameter metadata cannot identify whether the target is {@code TIMESTAMP} or {@code TIMESTAMP WITH TIME ZONE}? 4409 * 4410 * @return behavior to use when timestamp target metadata is unavailable or non-identifying 4411 * @since 4.2.0 4412 */ 4413 @NonNull 4414 public AmbiguousTimestampBindingStrategy getAmbiguousTimestampBindingStrategy() { 4415 return this.ambiguousTimestampBindingStrategy; 4416 } 4417 4418 /** 4419 * Gets the configured redactor used for non-secure parameters in diagnostics. 4420 * 4421 * @return the configured parameter redactor 4422 * @since 4.4.0 4423 */ 4424 @NonNull 4425 public ParameterRedactor getParameterRedactor() { 4426 return this.parameterRedactor; 4427 } 4428 4429 /** 4430 * Useful for single-shot "utility" calls that operate outside of normal query operations, e.g. pulling DB metadata. 4431 * <p> 4432 * Example: {@link #readDatabaseMetaData(DatabaseMetaDataReader)}. 4433 */ 4434 @NonNull 4435 <R> Optional<R> performRawConnectionOperation(@NonNull InternalRawConnectionOperation<R> rawConnectionOperation, 4436 @NonNull Boolean shouldParticipateInExistingTransactionIfPossible) { 4437 requireNonNull(rawConnectionOperation); 4438 requireNonNull(shouldParticipateInExistingTransactionIfPossible); 4439 4440 if (shouldParticipateInExistingTransactionIfPossible) { 4441 Optional<Transaction> transaction = currentTransactionForDatabaseOperation(); 4442 ReentrantLock connectionLock = transaction.isPresent() ? transaction.get().getConnectionLock() : null; 4443 // Try to participate in txn if it's available 4444 Connection connection = null; 4445 Throwable thrown = null; 4446 boolean connectionLockAcquired = false; 4447 4448 try { 4449 if (connectionLock != null) { 4450 lockInterruptibly(connectionLock, "use the transaction connection for a raw connection operation"); 4451 connectionLockAcquired = true; 4452 } 4453 4454 if (transaction.isPresent()) { 4455 connection = transaction.get().getConnection(); 4456 } else { 4457 try { 4458 connection = getDataSource().getConnection(); 4459 } catch (SQLException e) { 4460 throw new DatabaseException("Unable to acquire database connection", e); 4461 } 4462 } 4463 4464 return rawConnectionOperation.perform(connection); 4465 } catch (DatabaseException e) { 4466 thrown = e; 4467 throw e; 4468 } catch (Error e) { 4469 thrown = e; 4470 throw e; 4471 } catch (Exception e) { 4472 DatabaseException wrapped = databaseExceptionWithRawConnectionContext(connection, e); 4473 thrown = wrapped; 4474 throw wrapped; 4475 } finally { 4476 Throwable cleanupFailure = null; 4477 4478 try { 4479 // If this was a single-shot operation (not in a transaction), close the connection 4480 if (connection != null && !transaction.isPresent()) { 4481 try { 4482 closeConnection(connection); 4483 } catch (Throwable cleanupException) { 4484 cleanupFailure = cleanupException; 4485 } 4486 } 4487 } finally { 4488 if (connectionLockAcquired) 4489 connectionLock.unlock(); 4490 4491 if (cleanupFailure != null) { 4492 if (thrown != null) { 4493 thrown.addSuppressed(cleanupFailure); 4494 } else if (cleanupFailure instanceof RuntimeException) { 4495 throw (RuntimeException) cleanupFailure; 4496 } else if (cleanupFailure instanceof Error) { 4497 throw (Error) cleanupFailure; 4498 } else { 4499 throw new RuntimeException(cleanupFailure); 4500 } 4501 } 4502 } 4503 } 4504 } else { 4505 boolean acquiredConnection = false; 4506 Connection connection = null; 4507 Throwable thrown = null; 4508 4509 // Always get a fresh connection no matter what and close it afterwards 4510 try { 4511 connection = getDataSource().getConnection(); 4512 acquiredConnection = true; 4513 return rawConnectionOperation.perform(connection); 4514 } catch (DatabaseException e) { 4515 thrown = e; 4516 throw e; 4517 } catch (Error e) { 4518 thrown = e; 4519 throw e; 4520 } catch (Exception e) { 4521 DatabaseException wrapped = acquiredConnection 4522 ? databaseExceptionWithRawConnectionContext(connection, e) 4523 : new DatabaseException("Unable to acquire database connection", e); 4524 thrown = wrapped; 4525 throw wrapped; 4526 } finally { 4527 if (connection != null) { 4528 try { 4529 closeConnection(connection); 4530 } catch (Throwable cleanupException) { 4531 if (thrown != null) { 4532 thrown.addSuppressed(cleanupException); 4533 } else if (cleanupException instanceof RuntimeException) { 4534 throw (RuntimeException) cleanupException; 4535 } else if (cleanupException instanceof Error) { 4536 throw (Error) cleanupException; 4537 } else { 4538 throw new RuntimeException(cleanupException); 4539 } 4540 } 4541 } 4542 } 4543 } 4544 } 4545 4546 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4547 @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation, 4548 @NonNull DatabaseOperation databaseOperation) { 4549 performDatabaseOperation(statementContext, preparedStatementBindingOperation, databaseOperation, null); 4550 } 4551 4552 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4553 @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation, 4554 @NonNull DatabaseOperation databaseOperation, 4555 @Nullable Integer batchSize) { 4556 performDatabaseOperation(statementContext, preparedStatementBindingOperation, databaseOperation, batchSize, 4557 (connection, context) -> connection.prepareStatement(context.getStatement().getSql())); 4558 } 4559 4560 private <T> void performDatabaseOperation(@NonNull StatementContext<T> statementContext, 4561 @NonNull PreparedStatementBindingOperation preparedStatementBindingOperation, 4562 @NonNull DatabaseOperation databaseOperation, 4563 @Nullable Integer batchSize, 4564 @NonNull PreparedStatementFactory preparedStatementFactory) { 4565 requireNonNull(statementContext); 4566 requireNonNull(preparedStatementBindingOperation); 4567 requireNonNull(databaseOperation); 4568 requireNonNull(preparedStatementFactory); 4569 4570 long startTime = nanoTime(); 4571 Duration connectionAcquisitionDuration = null; 4572 Duration preparationDuration = null; 4573 Duration executionDuration = null; 4574 Duration resultSetMappingDuration = null; 4575 StatementResult statementResult = StatementResult.empty(); 4576 Exception exception = null; 4577 Throwable thrown = null; 4578 Connection connection = null; 4579 long connectionHeldStartTime = 0L; 4580 Optional<Transaction> transaction = currentTransactionForDatabaseOperation(); 4581 ReentrantLock connectionLock = transaction.isPresent() ? transaction.get().getConnectionLock() : null; 4582 boolean connectionLockAcquired = false; 4583 Connection previousDatabaseTypeDetectionConnection = null; 4584 boolean databaseTypeDetectionConnectionInstalled = false; 4585 4586 try { 4587 if (connectionLock != null) { 4588 lockInterruptibly(connectionLock, "execute a statement on the transaction connection"); 4589 connectionLockAcquired = true; 4590 } 4591 4592 boolean alreadyHasConnection = transaction.isPresent() && transaction.get().hasConnection(); 4593 if (transaction.isPresent()) { 4594 connection = transaction.get().getConnection(); 4595 } else { 4596 getMetricsCollectorDispatcher().willAcquireStatementConnection(statementContext); 4597 try { 4598 connection = getDataSource().getConnection(); 4599 } catch (SQLException e) { 4600 DatabaseException wrapped = new DatabaseException("Unable to acquire database connection", e); 4601 connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime); 4602 getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(statementContext, peekDatabaseType(), 4603 connectionAcquisitionDuration, wrapped); 4604 throw wrapped; 4605 } catch (RuntimeException e) { 4606 connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime); 4607 getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(statementContext, peekDatabaseType(), 4608 connectionAcquisitionDuration, e); 4609 throw e; 4610 } 4611 } 4612 4613 connectionAcquisitionDuration = alreadyHasConnection ? null : Duration.ofNanos(nanoTime() - startTime); 4614 previousDatabaseTypeDetectionConnection = this.databaseTypeDetectionConnectionHolder.get(); 4615 this.databaseTypeDetectionConnectionHolder.set(connection); 4616 databaseTypeDetectionConnectionInstalled = true; 4617 4618 if (!transaction.isPresent()) { 4619 connectionHeldStartTime = nanoTime(); 4620 Duration acquiredDuration = connectionAcquisitionDuration; 4621 MetricsCollectorDispatcher metricsCollectorDispatcher = getMetricsCollectorDispatcher(); 4622 if (metricsCollectorDispatcher.isEnabled()) 4623 dispatchWithDatabaseTypeDetectionConnection(connection, () -> 4624 metricsCollectorDispatcher.didAcquireStatementConnection(statementContext, acquiredDuration)); 4625 } 4626 startTime = nanoTime(); 4627 4628 try (PreparedStatement preparedStatement = preparedStatementFactory.prepare(connection, statementContext)) { 4629 preparedStatementBindingOperation.perform(preparedStatement); 4630 preparationDuration = Duration.ofNanos(nanoTime() - startTime); 4631 4632 getMetricsCollectorDispatcher().willExecuteStatement(statementContext); 4633 DatabaseOperationResult databaseOperationResult = databaseOperation.perform(preparedStatement); 4634 executionDuration = databaseOperationResult.getExecutionDuration().orElse(null); 4635 resultSetMappingDuration = databaseOperationResult.getResultSetMappingDuration().orElse(null); 4636 statementResult = databaseOperationResult.getStatementResult(); 4637 warmDatabaseTypeCacheForMetricsIfNeeded(statementContext); 4638 } 4639 } catch (DatabaseException e) { 4640 DatabaseException wrapped = databaseExceptionWithStatementContext(statementContext, e); 4641 exception = wrapped; 4642 thrown = wrapped; 4643 throw wrapped; 4644 } catch (Error e) { 4645 exception = databaseExceptionWithStatementContext(statementContext, e); 4646 thrown = e; 4647 throw e; 4648 } catch (Exception e) { 4649 DatabaseException wrapped = databaseExceptionWithStatementContext(statementContext, e); 4650 // Store the wrapped (scrubbed) form in the StatementLog, consistent with the DatabaseException 4651 // and Error paths - the raw driver exception remains available via the wrapped cause 4652 exception = wrapped; 4653 thrown = wrapped; 4654 throw wrapped; 4655 } finally { 4656 Throwable cleanupFailure = null; 4657 4658 if (databaseTypeDetectionConnectionInstalled) { 4659 if (previousDatabaseTypeDetectionConnection == null) 4660 this.databaseTypeDetectionConnectionHolder.remove(); 4661 else 4662 this.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection); 4663 } 4664 4665 try { 4666 cleanupFailure = closeStatementContextResources(statementContext, cleanupFailure); 4667 4668 // If this was a single-shot operation (not in a transaction), close the connection 4669 if (connection != null && !transaction.isPresent()) { 4670 Duration heldDuration = Duration.ofNanos(nanoTime() - connectionHeldStartTime); 4671 try { 4672 closeConnection(connection); 4673 getMetricsCollectorDispatcher().didReleaseStatementConnection(statementContext, heldDuration); 4674 } catch (Throwable cleanupException) { 4675 getMetricsCollectorDispatcher().didFailToReleaseStatementConnection(statementContext, heldDuration, cleanupException); 4676 if (cleanupFailure == null) 4677 cleanupFailure = cleanupException; 4678 else 4679 cleanupFailure.addSuppressed(cleanupException); 4680 } 4681 } 4682 } finally { 4683 if (connectionLockAcquired) 4684 connectionLock.unlock(); 4685 4686 StatementLog statementLog = 4687 StatementLog.withStatementContext(statementContext) 4688 .connectionAcquisitionDuration(connectionAcquisitionDuration) 4689 .preparationDuration(preparationDuration) 4690 .executionDuration(executionDuration) 4691 .resultSetMappingDuration(resultSetMappingDuration) 4692 .batchSize(batchSize) 4693 .exception(exception) 4694 .build(); 4695 4696 if (thrown == null && exception == null) { 4697 getMetricsCollectorDispatcher().didExecuteStatement(statementContext, statementLog, statementResult); 4698 } else { 4699 Throwable statementThrowable = thrown == null ? exception : thrown; 4700 getMetricsCollectorDispatcher().didFailToExecuteStatement(statementContext, statementLog, peekDatabaseType(), 4701 requireNonNull(statementThrowable)); 4702 } 4703 4704 try { 4705 getStatementLogger().log(statementLog); 4706 } catch (Throwable cleanupException) { 4707 if (cleanupFailure == null) 4708 cleanupFailure = cleanupException; 4709 else 4710 cleanupFailure.addSuppressed(cleanupException); 4711 } 4712 } 4713 4714 if (cleanupFailure != null) { 4715 if (thrown != null) { 4716 thrown.addSuppressed(cleanupFailure); 4717 } else if (cleanupFailure instanceof RuntimeException) { 4718 throw (RuntimeException) cleanupFailure; 4719 } else if (cleanupFailure instanceof Error) { 4720 throw (Error) cleanupFailure; 4721 } else { 4722 throw new RuntimeException(cleanupFailure); 4723 } 4724 } 4725 } 4726 } 4727 4728 @NonNull 4729 DataSource getDataSource() { 4730 return this.dataSource; 4731 } 4732 4733 @NonNull 4734 InstanceProvider getInstanceProvider() { 4735 return this.instanceProvider; 4736 } 4737 4738 @NonNull 4739 private PreparedStatementBinder getPreparedStatementBinder() { 4740 return this.preparedStatementBinder; 4741 } 4742 4743 /** 4744 * Resolves the effective {@link ResultSetMapper} for a statement: the per-query override carried on 4745 * the {@link StatementContext} when present, otherwise the database-wide instance. 4746 */ 4747 @NonNull 4748 private ResultSetMapper resultSetMapperFor(@NonNull StatementContext<?> statementContext) { 4749 requireNonNull(statementContext); 4750 ResultSetMapper resultSetMapperOverride = statementContext.getResultSetMapperOverride(); 4751 return resultSetMapperOverride != null ? resultSetMapperOverride : getResultSetMapper(); 4752 } 4753 4754 /** 4755 * Resolves the effective {@link PreparedStatementBinder} for a statement: the per-query override 4756 * carried on the {@link StatementContext} when present, otherwise the database-wide instance. 4757 */ 4758 @NonNull 4759 private PreparedStatementBinder preparedStatementBinderFor(@NonNull StatementContext<?> statementContext) { 4760 requireNonNull(statementContext); 4761 PreparedStatementBinder preparedStatementBinderOverride = statementContext.getPreparedStatementBinderOverride(); 4762 return preparedStatementBinderOverride != null ? preparedStatementBinderOverride : getPreparedStatementBinder(); 4763 } 4764 4765 @NonNull 4766 ResultSetMapper getResultSetMapper() { 4767 return this.resultSetMapper; 4768 } 4769 4770 @NonNull 4771 private StatementLogger getStatementLogger() { 4772 return this.statementLogger; 4773 } 4774 4775 @NonNull 4776 public MetricsCollector getMetricsCollector() { 4777 return getMetricsCollectorDispatcher().getMetricsCollector(); 4778 } 4779 4780 @NonNull 4781 MetricsCollectorDispatcher getMetricsCollectorDispatcher() { 4782 return this.metricsCollectorDispatcher; 4783 } 4784 4785 @NonNull 4786 private DatabaseOperationSupportStatus getExecuteLargeBatchSupported() { 4787 return this.executeLargeBatchSupported; 4788 } 4789 4790 private void setExecuteLargeBatchSupported(@NonNull DatabaseOperationSupportStatus executeLargeBatchSupported) { 4791 requireNonNull(executeLargeBatchSupported); 4792 this.executeLargeBatchSupported = executeLargeBatchSupported; 4793 } 4794 4795 @NonNull 4796 private DatabaseOperationSupportStatus getExecuteLargeUpdateSupported() { 4797 return this.executeLargeUpdateSupported; 4798 } 4799 4800 private void setExecuteLargeUpdateSupported(@NonNull DatabaseOperationSupportStatus executeLargeUpdateSupported) { 4801 requireNonNull(executeLargeUpdateSupported); 4802 this.executeLargeUpdateSupported = executeLargeUpdateSupported; 4803 } 4804 4805 @NonNull 4806 Object generateId() { 4807 // "Unique" keys 4808 return this.defaultIdGenerator.incrementAndGet(); 4809 } 4810 4811 @FunctionalInterface 4812 private interface DatabaseOperation { 4813 @NonNull 4814 DatabaseOperationResult perform(@NonNull PreparedStatement preparedStatement) throws Exception; 4815 } 4816 4817 @FunctionalInterface 4818 private interface PreparedStatementFactory { 4819 @NonNull 4820 PreparedStatement prepare(@NonNull Connection connection, 4821 @NonNull StatementContext<?> statementContext) throws SQLException; 4822 } 4823 4824 @FunctionalInterface 4825 private interface PreparedStatementBindingOperation { 4826 void perform(@NonNull PreparedStatement preparedStatement) throws Exception; 4827 } 4828 4829 @NotThreadSafe 4830 private static final class StreamingResultSet<T> implements java.util.Iterator<T>, AutoCloseable { 4831 private final Database database; 4832 private final StatementContext<T> statementContext; 4833 private final List<Object> parameters; 4834 @Nullable 4835 private final PreparedStatementCustomizer preparedStatementCustomizer; 4836 @NonNull 4837 private final Optional<Transaction> transaction; 4838 @Nullable 4839 private final ReentrantLock connectionLock; 4840 @Nullable 4841 private Connection connection; 4842 @Nullable 4843 private PreparedStatement preparedStatement; 4844 @Nullable 4845 private ResultSet resultSet; 4846 private boolean closed; 4847 private boolean hasNextEvaluated; 4848 private boolean hasNext; 4849 @Nullable 4850 private Duration connectionAcquisitionDuration; 4851 @Nullable 4852 private Duration preparationDuration; 4853 @Nullable 4854 private Duration executionDuration; 4855 private long resultSetMappingNanos; 4856 @Nullable 4857 private Exception exception; 4858 @Nullable 4859 private Throwable thrown; 4860 private long rowsConsumed; 4861 private long openStartTime; 4862 private boolean exhausted; 4863 private boolean openFailed; 4864 private boolean terminalMetricsEmitted; 4865 @Nullable 4866 private Throwable callbackThrowable; 4867 @Nullable 4868 private Throwable iterationThrowable; 4869 @Nullable 4870 private Throwable cleanupFailure; 4871 private long connectionHeldStartTime; 4872 @NonNull 4873 private DatabaseDialect databaseStreamDialect = GenericDialect.INSTANCE; 4874 @NonNull 4875 private DatabaseStreamState databaseStreamState = DatabaseStreamState.none(); 4876 private boolean connectionLockAcquired; 4877 private final boolean queryFetchSizeConfigured; 4878 4879 private StreamingResultSet(@NonNull Database database, 4880 @NonNull StatementContext<T> statementContext, 4881 @NonNull List<Object> parameters, 4882 @Nullable PreparedStatementCustomizer preparedStatementCustomizer, 4883 boolean queryFetchSizeConfigured) { 4884 this.database = requireNonNull(database); 4885 this.statementContext = requireNonNull(statementContext); 4886 this.parameters = requireNonNull(parameters); 4887 this.preparedStatementCustomizer = preparedStatementCustomizer; 4888 this.transaction = database.currentTransactionForDatabaseOperation(); 4889 this.connectionLock = this.transaction.isPresent() ? this.transaction.get().getConnectionLock() : null; 4890 this.queryFetchSizeConfigured = queryFetchSizeConfigured; 4891 4892 open(); 4893 } 4894 4895 private void open() { 4896 long startTime = nanoTime(); 4897 this.openStartTime = startTime; 4898 this.database.getMetricsCollectorDispatcher().willOpenStream(this.statementContext); 4899 Connection previousDatabaseTypeDetectionConnection = null; 4900 boolean databaseTypeDetectionConnectionInstalled = false; 4901 4902 try { 4903 if (this.connectionLock != null) { 4904 lockInterruptibly(this.connectionLock, "open a stream on the transaction connection"); 4905 this.connectionLockAcquired = true; 4906 } 4907 4908 boolean alreadyHasConnection = this.transaction.isPresent() && this.transaction.get().hasConnection(); 4909 if (this.transaction.isPresent()) { 4910 this.connection = this.transaction.get().getConnection(); 4911 } else { 4912 this.database.getMetricsCollectorDispatcher().willAcquireStatementConnection(this.statementContext); 4913 try { 4914 this.connection = this.database.getDataSource().getConnection(); 4915 } catch (SQLException e) { 4916 DatabaseException wrapped = new DatabaseException("Unable to acquire database connection", e); 4917 this.connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime); 4918 this.database.getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(this.statementContext, 4919 this.database.peekDatabaseType(), this.connectionAcquisitionDuration, wrapped); 4920 throw wrapped; 4921 } catch (RuntimeException e) { 4922 this.connectionAcquisitionDuration = Duration.ofNanos(nanoTime() - startTime); 4923 this.database.getMetricsCollectorDispatcher().didFailToAcquireStatementConnection(this.statementContext, 4924 this.database.peekDatabaseType(), this.connectionAcquisitionDuration, e); 4925 throw e; 4926 } 4927 } 4928 this.connectionAcquisitionDuration = alreadyHasConnection ? null : Duration.ofNanos(nanoTime() - startTime); 4929 previousDatabaseTypeDetectionConnection = this.database.databaseTypeDetectionConnectionHolder.get(); 4930 this.database.databaseTypeDetectionConnectionHolder.set(this.connection); 4931 databaseTypeDetectionConnectionInstalled = true; 4932 4933 if (this.transaction.isEmpty()) { 4934 this.connectionHeldStartTime = nanoTime(); 4935 MetricsCollectorDispatcher metricsCollectorDispatcher = this.database.getMetricsCollectorDispatcher(); 4936 if (metricsCollectorDispatcher.isEnabled()) 4937 this.database.dispatchWithDatabaseTypeDetectionConnection(requireNonNull(this.connection), () -> 4938 metricsCollectorDispatcher.didAcquireStatementConnection(this.statementContext, this.connectionAcquisitionDuration)); 4939 } 4940 startTime = nanoTime(); 4941 4942 this.databaseStreamDialect = databaseDialectForStreamingConnection(); 4943 this.databaseStreamState = this.databaseStreamDialect.configureStreamingConnection(requireNonNull(this.connection), this.transaction.isPresent()); 4944 4945 this.preparedStatement = this.databaseStreamDialect.prepareStreamingStatement(requireNonNull(this.connection), this.statementContext); 4946 this.database.applyPreparedStatementCustomizer(this.statementContext, this.preparedStatement, this.preparedStatementCustomizer); 4947 this.databaseStreamDialect.configureStreamingPreparedStatement(this.preparedStatement, this.databaseStreamState, 4948 this.transaction.isPresent(), this.queryFetchSizeConfigured); 4949 if (this.parameters.size() > 0) 4950 this.database.performPreparedStatementBinding(this.statementContext, this.preparedStatement, this.parameters); 4951 this.preparationDuration = Duration.ofNanos(nanoTime() - startTime); 4952 4953 startTime = nanoTime(); 4954 this.resultSet = this.preparedStatement.executeQuery(); 4955 this.executionDuration = Duration.ofNanos(nanoTime() - startTime); 4956 this.database.warmDatabaseTypeCacheForMetricsIfNeeded(this.statementContext); 4957 this.database.getMetricsCollectorDispatcher().didOpenStream(this.statementContext, Duration.ofNanos(nanoTime() - this.openStartTime)); 4958 } catch (DatabaseException e) { 4959 DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e); 4960 this.exception = wrapped; 4961 this.thrown = wrapped; 4962 this.openFailed = true; 4963 this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(), 4964 Duration.ofNanos(nanoTime() - this.openStartTime), wrapped); 4965 close(); 4966 throw wrapped; 4967 } catch (Exception e) { 4968 DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e); 4969 this.exception = wrapped; 4970 this.thrown = wrapped; 4971 this.openFailed = true; 4972 this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(), 4973 Duration.ofNanos(nanoTime() - this.openStartTime), wrapped); 4974 close(); 4975 throw wrapped; 4976 } catch (Error e) { 4977 this.exception = databaseExceptionWithStatementContext(this.statementContext, e); 4978 this.thrown = e; 4979 this.openFailed = true; 4980 this.database.getMetricsCollectorDispatcher().didFailToOpenStream(this.statementContext, this.database.peekDatabaseType(), 4981 Duration.ofNanos(nanoTime() - this.openStartTime), e); 4982 close(); 4983 throw e; 4984 } finally { 4985 if (databaseTypeDetectionConnectionInstalled) { 4986 if (previousDatabaseTypeDetectionConnection == null) 4987 this.database.databaseTypeDetectionConnectionHolder.remove(); 4988 else 4989 this.database.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection); 4990 } 4991 } 4992 } 4993 4994 @NonNull 4995 private DatabaseDialect databaseDialectForStreamingConnection() { 4996 try { 4997 return this.database.getDatabaseDialect(requireNonNull(this.connection)); 4998 } catch (DatabaseException e) { 4999 return this.database.peekDatabaseType().dialect(); 5000 } 5001 } 5002 5003 @Override 5004 public boolean hasNext() { 5005 if (this.closed) 5006 return false; 5007 5008 if (!this.hasNextEvaluated) { 5009 try { 5010 this.hasNext = this.resultSet != null && this.resultSet.next(); 5011 this.hasNextEvaluated = true; 5012 if (!this.hasNext) { 5013 this.exhausted = true; 5014 close(); 5015 } 5016 } catch (SQLException e) { 5017 DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e); 5018 this.exception = wrapped; 5019 this.thrown = wrapped; 5020 this.iterationThrowable = wrapped; 5021 close(); 5022 throw wrapped; 5023 } catch (RuntimeException e) { 5024 this.exception = e; 5025 this.thrown = e; 5026 this.iterationThrowable = e; 5027 close(); 5028 throw e; 5029 } catch (Error e) { 5030 this.exception = databaseExceptionWithStatementContext(this.statementContext, e); 5031 this.thrown = e; 5032 this.iterationThrowable = e; 5033 close(); 5034 throw e; 5035 } 5036 } 5037 5038 return this.hasNext; 5039 } 5040 5041 @Override 5042 public T next() { 5043 if (!hasNext()) 5044 throw new java.util.NoSuchElementException(); 5045 5046 this.hasNextEvaluated = false; 5047 long startTime = nanoTime(); 5048 Connection previousDatabaseTypeDetectionConnection = this.database.databaseTypeDetectionConnectionHolder.get(); 5049 this.database.databaseTypeDetectionConnectionHolder.set(requireNonNull(this.connection)); 5050 5051 try { 5052 T value = this.database.resultSetMapperFor(this.statementContext) 5053 .map(this.statementContext, requireNonNull(this.resultSet), this.statementContext.getResultSetRowType().get(), this.database.getInstanceProvider()) 5054 .orElse(null); 5055 5056 this.resultSetMappingNanos += nanoTime() - startTime; 5057 this.rowsConsumed++; 5058 return value; 5059 } catch (SQLException e) { 5060 DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, 5061 format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), this.statementContext.getResultSetRowType().get()), e); 5062 this.exception = wrapped; 5063 this.thrown = wrapped; 5064 this.iterationThrowable = wrapped; 5065 close(); 5066 throw wrapped; 5067 } catch (DatabaseException e) { 5068 DatabaseException wrapped = databaseExceptionWithStatementContext(this.statementContext, e); 5069 this.exception = wrapped; 5070 this.thrown = wrapped; 5071 this.iterationThrowable = wrapped; 5072 close(); 5073 throw wrapped; 5074 } catch (RuntimeException e) { 5075 this.exception = e; 5076 this.thrown = e; 5077 this.iterationThrowable = e; 5078 close(); 5079 throw e; 5080 } catch (Error e) { 5081 this.exception = databaseExceptionWithStatementContext(this.statementContext, 5082 format("Unable to map JDBC %s row to %s", ResultSet.class.getSimpleName(), this.statementContext.getResultSetRowType().get()), e); 5083 this.thrown = e; 5084 this.iterationThrowable = e; 5085 close(); 5086 throw e; 5087 } finally { 5088 if (previousDatabaseTypeDetectionConnection == null) 5089 this.database.databaseTypeDetectionConnectionHolder.remove(); 5090 else 5091 this.database.databaseTypeDetectionConnectionHolder.set(previousDatabaseTypeDetectionConnection); 5092 } 5093 } 5094 5095 private void callbackFailed(@NonNull Throwable throwable) { 5096 requireNonNull(throwable); 5097 this.callbackThrowable = throwable; 5098 } 5099 5100 @Override 5101 public void close() { 5102 if (this.closed) 5103 return; 5104 5105 if (this.connectionLockAcquired && this.connectionLock != null && !this.connectionLock.isHeldByCurrentThread()) 5106 throw new DatabaseException("Transactional streams must be closed by the thread that opened them"); 5107 5108 this.closed = true; 5109 Throwable cleanupFailure = null; 5110 5111 try { 5112 cleanupFailure = closeStatementContextResources(this.statementContext, cleanupFailure); 5113 5114 if (this.resultSet != null) { 5115 try { 5116 this.resultSet.close(); 5117 } catch (Throwable cleanupException) { 5118 cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException); 5119 } 5120 } 5121 5122 if (this.preparedStatement != null) { 5123 try { 5124 this.preparedStatement.close(); 5125 } catch (Throwable cleanupException) { 5126 cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException); 5127 } 5128 } 5129 5130 cleanupFailure = completeDialectStreamingConnectionIfNeeded(cleanupFailure); 5131 5132 if (this.connection != null && this.transaction.isEmpty()) { 5133 Duration heldDuration = this.connectionHeldStartTime == 0L 5134 ? Duration.ZERO 5135 : Duration.ofNanos(nanoTime() - this.connectionHeldStartTime); 5136 try { 5137 this.database.closeConnection(this.connection); 5138 this.database.getMetricsCollectorDispatcher().didReleaseStatementConnection(this.statementContext, heldDuration); 5139 } catch (Throwable cleanupException) { 5140 this.database.getMetricsCollectorDispatcher().didFailToReleaseStatementConnection(this.statementContext, heldDuration, cleanupException); 5141 cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException); 5142 } 5143 } 5144 } finally { 5145 if (this.connectionLockAcquired) { 5146 this.connectionLock.unlock(); 5147 this.connectionLockAcquired = false; 5148 } 5149 5150 Duration mappingDuration = this.resultSetMappingNanos == 0L ? null : Duration.ofNanos(this.resultSetMappingNanos); 5151 5152 StatementLog statementLog = 5153 StatementLog.withStatementContext(this.statementContext) 5154 .connectionAcquisitionDuration(this.connectionAcquisitionDuration) 5155 .preparationDuration(this.preparationDuration) 5156 .executionDuration(this.executionDuration) 5157 .resultSetMappingDuration(mappingDuration) 5158 .exception(this.exception) 5159 .build(); 5160 5161 if (this.thrown == null && this.exception == null) { 5162 this.database.getMetricsCollectorDispatcher().didExecuteStatement(this.statementContext, statementLog, StatementResult.empty()); 5163 } else { 5164 Throwable statementThrowable = this.thrown == null ? this.exception : this.thrown; 5165 this.database.getMetricsCollectorDispatcher().didFailToExecuteStatement(this.statementContext, statementLog, 5166 this.database.peekDatabaseType(), requireNonNull(statementThrowable)); 5167 } 5168 5169 try { 5170 this.database.getStatementLogger().log(statementLog); 5171 } catch (Throwable cleanupException) { 5172 cleanupFailure = cleanupFailure == null ? cleanupException : addSuppressed(cleanupFailure, cleanupException); 5173 } 5174 } 5175 5176 this.cleanupFailure = cleanupFailure; 5177 5178 if (cleanupFailure != null) { 5179 if (this.thrown != null) { 5180 this.thrown.addSuppressed(cleanupFailure); 5181 } else if (cleanupFailure instanceof RuntimeException) { 5182 throw (RuntimeException) cleanupFailure; 5183 } else if (cleanupFailure instanceof Error) { 5184 throw (Error) cleanupFailure; 5185 } else { 5186 throw new RuntimeException(cleanupFailure); 5187 } 5188 } 5189 } 5190 5191 @Nullable 5192 private Throwable completeDialectStreamingConnectionIfNeeded(@Nullable Throwable cleanupFailure) { 5193 if (this.connection == null) 5194 return cleanupFailure; 5195 5196 boolean streamSucceeded = this.thrown == null && this.exception == null && this.callbackThrowable == null && !this.openFailed; 5197 cleanupFailure = this.databaseStreamDialect.completeStreamingConnection(requireNonNull(this.connection), 5198 this.databaseStreamState, streamSucceeded, cleanupFailure); 5199 5200 this.databaseStreamDialect = GenericDialect.INSTANCE; 5201 this.databaseStreamState = DatabaseStreamState.none(); 5202 return cleanupFailure; 5203 } 5204 5205 private void emitTerminalMetrics() { 5206 if (this.terminalMetricsEmitted) 5207 return; 5208 5209 this.terminalMetricsEmitted = true; 5210 5211 if (this.openFailed) 5212 return; 5213 5214 MetricsCollector.StreamTerminalOutcome outcome; 5215 Throwable throwable; 5216 5217 if (this.iterationThrowable != null) { 5218 outcome = MetricsCollector.StreamTerminalOutcome.ITERATION_FAILURE; 5219 throwable = this.iterationThrowable; 5220 } else if (this.callbackThrowable != null) { 5221 outcome = MetricsCollector.StreamTerminalOutcome.CALLBACK_FAILURE; 5222 throwable = this.callbackThrowable; 5223 } else if (this.exhausted) { 5224 outcome = MetricsCollector.StreamTerminalOutcome.COMPLETED_NORMALLY; 5225 throwable = this.cleanupFailure; 5226 } else { 5227 outcome = MetricsCollector.StreamTerminalOutcome.EARLY_CLOSE; 5228 throwable = this.cleanupFailure; 5229 } 5230 5231 this.database.getMetricsCollectorDispatcher().didCloseStream(this.statementContext, outcome, this.rowsConsumed, 5232 Duration.ofNanos(nanoTime() - this.openStartTime), throwable); 5233 } 5234 5235 @NonNull 5236 private static Throwable addSuppressed(@NonNull Throwable existing, 5237 @NonNull Throwable additional) { 5238 existing.addSuppressed(additional); 5239 return existing; 5240 } 5241 } 5242 5243 /** 5244 * Builder used to construct instances of {@link Database}. 5245 * <p> 5246 * This class is intended for use by a single thread. 5247 * 5248 * @author <a href="https://www.revetkn.com">Mark Allen</a> 5249 * @since 1.0.0 5250 */ 5251 @NotThreadSafe 5252 public static class Builder { 5253 @NonNull 5254 private final DataSource dataSource; 5255 @Nullable 5256 private DatabaseType databaseType; 5257 @Nullable 5258 private ZoneId timeZone; 5259 @Nullable 5260 private AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy; 5261 @Nullable 5262 private InstanceProvider instanceProvider; 5263 @Nullable 5264 private PreparedStatementBinder preparedStatementBinder; 5265 @Nullable 5266 private ResultSetMapper resultSetMapper; 5267 @Nullable 5268 private StatementLogger statementLogger; 5269 @Nullable 5270 private ParameterRedactor parameterRedactor; 5271 @Nullable 5272 private MetricsCollector metricsCollector; 5273 @Nullable 5274 private Duration queryTimeout; 5275 @Nullable 5276 private Integer fetchSize; 5277 @Nullable 5278 private Integer maxRows; 5279 @Nullable 5280 private Integer parsedSqlCacheCapacity; 5281 5282 private Builder(@NonNull DataSource dataSource) { 5283 this.dataSource = requireNonNull(dataSource); 5284 this.databaseType = null; 5285 this.metricsCollector = null; 5286 this.queryTimeout = null; 5287 this.fetchSize = null; 5288 this.maxRows = null; 5289 this.parsedSqlCacheCapacity = null; 5290 } 5291 5292 /** 5293 * Overrides automatic database type detection. 5294 * <p> 5295 * If {@code null}, the database type is detected lazily when database-type-specific behavior is first needed. 5296 * Supplying a non-null value avoids automatic detection and its metadata lookup entirely. 5297 * 5298 * @param databaseType the database type to use (null to enable auto-detection) 5299 * @return this {@code Builder}, for chaining 5300 * @since 4.0.0 5301 */ 5302 @NonNull 5303 public Builder databaseType(@Nullable DatabaseType databaseType) { 5304 this.databaseType = databaseType; 5305 return this; 5306 } 5307 5308 /** 5309 * Configures the database time zone Pyranid should use when converting zone-less temporal values. 5310 * <p> 5311 * This value is used when mapping {@code TIMESTAMP} values to instant-based Java types, and when binding 5312 * {@link java.time.Instant} or {@link java.time.OffsetDateTime} parameters to known {@code TIMESTAMP} 5313 * targets. It also applies to ambiguous timestamp bindings if 5314 * {@link #ambiguousTimestampBindingStrategy(AmbiguousTimestampBindingStrategy)} is configured with 5315 * {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITHOUT_TIME_ZONE}. 5316 * <p> 5317 * If {@code null}, Pyranid uses {@link ZoneId#systemDefault()}. 5318 * 5319 * @param timeZone database time zone to use, or {@code null} for the JVM default zone 5320 * @return this {@code Builder}, for chaining 5321 * @since 3.0.0 5322 */ 5323 @NonNull 5324 public Builder timeZone(@Nullable ZoneId timeZone) { 5325 this.timeZone = timeZone; 5326 return this; 5327 } 5328 5329 /** 5330 * Configures how Pyranid binds {@link java.time.Instant} and {@link java.time.OffsetDateTime} parameters 5331 * when JDBC parameter metadata cannot identify whether the target is {@code TIMESTAMP} or 5332 * {@code TIMESTAMP WITH TIME ZONE}. 5333 * <p> 5334 * The default, {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITH_TIME_ZONE}, is appropriate for 5335 * timestamp-with-time-zone targets. Use 5336 * {@link AmbiguousTimestampBindingStrategy#TIMESTAMP_WITHOUT_TIME_ZONE} for drivers or proxies that 5337 * cannot provide identifying parameter metadata when your target columns are zone-less {@code TIMESTAMP} 5338 * values that should be interpreted in {@link #timeZone(ZoneId)}. 5339 * 5340 * @param ambiguousTimestampBindingStrategy strategy to use, or {@code null} for the default 5341 * @return this {@code Builder}, for chaining 5342 * @since 4.2.0 5343 */ 5344 @NonNull 5345 public Builder ambiguousTimestampBindingStrategy(@Nullable AmbiguousTimestampBindingStrategy ambiguousTimestampBindingStrategy) { 5346 this.ambiguousTimestampBindingStrategy = ambiguousTimestampBindingStrategy; 5347 return this; 5348 } 5349 5350 @NonNull 5351 public Builder instanceProvider(@Nullable InstanceProvider instanceProvider) { 5352 this.instanceProvider = instanceProvider; 5353 return this; 5354 } 5355 5356 @NonNull 5357 public Builder preparedStatementBinder(@Nullable PreparedStatementBinder preparedStatementBinder) { 5358 this.preparedStatementBinder = preparedStatementBinder; 5359 return this; 5360 } 5361 5362 @NonNull 5363 public Builder resultSetMapper(@Nullable ResultSetMapper resultSetMapper) { 5364 this.resultSetMapper = resultSetMapper; 5365 return this; 5366 } 5367 5368 /** 5369 * Configures the statement logger for the {@link Database} being built. 5370 * <p> 5371 * {@link StatementLogger} failures are fail-fast: a logger exception can make a successful statement operation throw, 5372 * and inside a Pyranid transaction it participates in normal rollback handling. If the database statement itself failed, 5373 * logger failures are suppressed onto the primary statement failure. 5374 * 5375 * @param statementLogger statement logger to use, or {@code null} for a no-op logger 5376 * @return this {@code Builder}, for chaining 5377 */ 5378 @NonNull 5379 public Builder statementLogger(@Nullable StatementLogger statementLogger) { 5380 this.statementLogger = statementLogger; 5381 return this; 5382 } 5383 5384 /** 5385 * Configures the redactor used for non-secure parameters in diagnostics. 5386 * <p> 5387 * {@link SecureParameter} values always render via {@link SecureParameter#getMask()} and are never passed to this 5388 * redactor. Batch executions render a bounded batch summary instead of invoking the redactor for each batch value. 5389 * Specify {@code null} or omit this setter to render non-secure, non-batch values verbatim. 5390 * 5391 * @param parameterRedactor parameter redactor to use, or {@code null} for the default 5392 * @return this {@code Builder}, for chaining 5393 * @since 4.4.0 5394 */ 5395 @NonNull 5396 public Builder parameterRedactor(@Nullable ParameterRedactor parameterRedactor) { 5397 this.parameterRedactor = parameterRedactor; 5398 return this; 5399 } 5400 5401 /** 5402 * Configures the metrics collector for the {@link Database} being built. 5403 * <p> 5404 * Like all {@code Database} configuration, this value is fixed at {@link #build()} time. Specify {@code null} 5405 * or omit this setter to disable metrics collection. 5406 * 5407 * @param metricsCollector metrics collector to use, or {@code null} to disable 5408 * @return this {@code Builder}, for chaining 5409 * @since 4.2.0 5410 */ 5411 @NonNull 5412 public Builder metricsCollector(@Nullable MetricsCollector metricsCollector) { 5413 this.metricsCollector = metricsCollector; 5414 return this; 5415 } 5416 5417 /** 5418 * Configures a database-wide JDBC query timeout default. 5419 * <p> 5420 * This maps to {@link java.sql.Statement#setQueryTimeout(int)}. {@code null} leaves the timeout unset. 5421 * {@link Duration#ZERO} disables the JDBC timeout. Positive sub-second durations are rounded up to one second 5422 * because JDBC accepts whole seconds. Per-query {@link Query#queryTimeout(Duration)} settings override this value, 5423 * and {@link Query#customize(PreparedStatementCustomizer)} can override both settings before parameter binding. 5424 * 5425 * @param queryTimeout timeout to apply by default, or {@code null} to leave unset 5426 * @return this {@code Builder}, for chaining 5427 * @since 4.2.0 5428 */ 5429 @NonNull 5430 public Builder queryTimeout(@Nullable Duration queryTimeout) { 5431 this.queryTimeout = validateQueryTimeout(queryTimeout); 5432 return this; 5433 } 5434 5435 /** 5436 * Configures a database-wide JDBC fetch size default. 5437 * <p> 5438 * This maps to {@link java.sql.Statement#setFetchSize(int)}. {@code null} leaves the fetch size unset. A value 5439 * of {@code 0} uses the driver's default fetch-size behavior. Per-query {@link Query#fetchSize(Integer)} 5440 * settings override this value, and {@link Query#customize(PreparedStatementCustomizer)} can override both 5441 * settings before parameter binding. 5442 * 5443 * @param fetchSize fetch size to apply by default, or {@code null} to leave unset 5444 * @return this {@code Builder}, for chaining 5445 * @since 4.2.0 5446 */ 5447 @NonNull 5448 public Builder fetchSize(@Nullable Integer fetchSize) { 5449 this.fetchSize = validateNonNegativeStatementSetting("fetchSize", fetchSize); 5450 return this; 5451 } 5452 5453 /** 5454 * Configures a database-wide JDBC maximum row count default. 5455 * <p> 5456 * This maps to {@link java.sql.Statement#setMaxRows(int)}. {@code null} leaves the maximum row count unset. A 5457 * value of {@code 0} disables the JDBC row limit. Per-query {@link Query#maxRows(Integer)} settings override 5458 * this value, and {@link Query#customize(PreparedStatementCustomizer)} can override both settings before parameter 5459 * binding. 5460 * 5461 * @param maxRows maximum rows to apply by default, or {@code null} to leave unset 5462 * @return this {@code Builder}, for chaining 5463 * @since 4.2.0 5464 */ 5465 @NonNull 5466 public Builder maxRows(@Nullable Integer maxRows) { 5467 this.maxRows = validateNonNegativeStatementSetting("maxRows", maxRows); 5468 return this; 5469 } 5470 5471 /** 5472 * Configures the size of the parsed SQL cache. 5473 * <p> 5474 * A value of {@code 0} disables caching. A value of {@code null} uses the default size. 5475 * 5476 * @param parsedSqlCacheCapacity cache size (0 disables caching, null uses default) 5477 * @return this {@code Builder}, for chaining 5478 */ 5479 @NonNull 5480 public Builder parsedSqlCacheCapacity(@Nullable Integer parsedSqlCacheCapacity) { 5481 if (parsedSqlCacheCapacity != null && parsedSqlCacheCapacity < 0) 5482 throw new IllegalArgumentException("parsedSqlCacheCapacity must be >= 0"); 5483 5484 this.parsedSqlCacheCapacity = parsedSqlCacheCapacity; 5485 return this; 5486 } 5487 5488 @NonNull 5489 public Database build() { 5490 return new Database(this); 5491 } 5492 } 5493 5494 @ThreadSafe 5495 static class DatabaseOperationResult { 5496 @Nullable 5497 private final Duration executionDuration; 5498 @Nullable 5499 private final Duration resultSetMappingDuration; 5500 @NonNull 5501 private final StatementResult statementResult; 5502 5503 public DatabaseOperationResult(@Nullable Duration executionDuration, 5504 @Nullable Duration resultSetMappingDuration) { 5505 this(executionDuration, resultSetMappingDuration, StatementResult.empty()); 5506 } 5507 5508 public DatabaseOperationResult(@Nullable Duration executionDuration, 5509 @Nullable Duration resultSetMappingDuration, 5510 @NonNull StatementResult statementResult) { 5511 this.executionDuration = executionDuration; 5512 this.resultSetMappingDuration = resultSetMappingDuration; 5513 this.statementResult = requireNonNull(statementResult); 5514 } 5515 5516 @NonNull 5517 public Optional<Duration> getExecutionDuration() { 5518 return Optional.ofNullable(this.executionDuration); 5519 } 5520 5521 @NonNull 5522 public Optional<Duration> getResultSetMappingDuration() { 5523 return Optional.ofNullable(this.resultSetMappingDuration); 5524 } 5525 5526 @NonNull 5527 public StatementResult getStatementResult() { 5528 return this.statementResult; 5529 } 5530 } 5531 5532 @NotThreadSafe 5533 static class ResultHolder<T> { 5534 T value; 5535 } 5536 5537 enum DatabaseOperationSupportStatus { 5538 UNKNOWN, 5539 YES, 5540 NO 5541 } 5542 5543}