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