Oracle Warning Segments

.github/workflows/startup/01_createUser.sql

@@ -36,12 +36,13 @@
 --   v$transaction (to verify if TransactionDefinitions are applied).
 --   v$session (to verify if VSESSION_* Options are applied).
 ALTER SESSION SET CONTAINER=xepdb1;
-CREATE ROLE r2dbc_test_user;
-GRANT SELECT ON v_$open_cursor TO r2dbc_test_user;
-GRANT SELECT ON v_$transaction TO r2dbc_test_user;
-GRANT SELECT ON v_$session TO r2dbc_test_user;
+CREATE ROLE r2dbc_test_role;
+GRANT SELECT ON v_$open_cursor TO r2dbc_test_role;
+GRANT SELECT ON v_$transaction TO r2dbc_test_role;
+GRANT SELECT ON v_$session TO r2dbc_test_role;
+GRANT CREATE VIEW TO r2dbc_test_role;
 
 CREATE USER test IDENTIFIED BY test;
-GRANT connect, resource, unlimited tablespace, r2dbc_test_user TO test;
+GRANT connect, resource, unlimited tablespace, r2dbc_test_role TO test;
 ALTER USER test DEFAULT TABLESPACE users;
 ALTER USER test TEMPORARY TABLESPACE temp;

README.md

@@ -344,12 +344,32 @@ Oracle R2DBC's implementation of Publishers that emit multiple items will
 typically defer execution until a Subscriber signals demand, and not support 
 multiple subscribers.
 
-### Errors
+### Errors and Warnings
 Oracle R2DBC creates R2dbcExceptions having the same ORA-XXXXX error codes 
-used by Oracle Database and Oracle JDBC.
+used by Oracle Database and Oracle JDBC. The
+[Database Error Messages](https://docs.oracle.com/en/database/oracle/oracle-database/21/errmg/ORA-00000.html#GUID-27437B7F-F0C3-4F1F-9C6E-6780706FB0F6)
+document provides a reference for all ORA-XXXXX error codes.
 
-A reference for the ORA-XXXXX error codes can be found 
-[here](https://docs.oracle.com/en/database/oracle/oracle-database/21/errmg/ORA-00000.html#GUID-27437B7F-F0C3-4F1F-9C6E-6780706FB0F6)
+Warning messages from Oracle Database are emitted as 
+`oracle.r2dbc.OracleR2dbcWarning` segments. These segments may be consumed using
+`Result.flatMap(Function)`:
+```java
+result.flatMap(segment -> {
+  if (segment instanceof OracleR2dbcWarning) {
+    logWarning(((OracleR2dbcWarning)segment).getMessage());
+    return emptyPublisher();
+  }
+  else if (segment instanceof Result.Message){
+    ... handle an error ...
+  }
+  else {
+    ... handle other segment types ...
+  }
+})
+```
+Unlike the errors of standard `Result.Message` segments, if a warning is not
+consumed by `flatMap`, then it will be silently discarded when a `Result` is 
+consumed using the `map` or `getRowsUpdated` methods.
 
 ### Transactions
 Oracle R2DBC uses READ COMMITTED as the default transaction isolation level.

src/main/java/oracle/r2dbc/OracleR2dbcWarning.java

@@ -0,0 +1,57 @@
+package oracle.r2dbc;
+
+import io.r2dbc.spi.Result;
+
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+/**
+ * <p>
+ * A subtype of {@link Result.Message} that provides information on warnings
+ * raised by Oracle Database.
+ * </p><p>
+ * When a SQL command results in a warning, Oracle R2DBC emits a {@link Result}
+ * with an {@code OracleR2dbcWarning} segment in addition to any other segments
+ * that resulted from the SQL command. For example, if a SQL {@code SELECT}
+ * command results in a warning, then an {@code OracleR2dbcWarning} segment is
+ * included with the result, along with any {@link Result.RowSegment}s returned
+ * by the {@code SELECT}.
+ * </p><p>
+ * R2DBC drivers typically emit {@code onError} signals for {@code Message}
+ * segments that are not consumed by {@link Result#filter(Predicate)} or
+ * {@link Result#flatMap(Function)}. Oracle R2DBC does not apply this behavior
+ * for warning messages. If an {@code OracleR2dbcWarning}
+ * segment is not consumed by the {@code filter} or {@code flatMap} methods of
+ * a {@code Result}, then the warning is discarded and the result may be
+ * consumed as normal with with the {@code map} or {@code getRowsUpdated}
+ * methods.
+ * </p><p>
+ * Warning messages may be consumed with {@link Result#flatMap(Function)}:
+ * </p><pre>{@code
+ * result.flatMap(segment -> {
+ *   if (segment instanceof OracleR2dbcWarning) {
+ *     logWarning(((OracleR2dbcWarning)segment).getMessage());
+ *     return emptyPublisher();
+ *   }
+ *   else {
+ *     ... handle other segment types ...
+ *   }
+ * })
+ * }</pre><p>
+ * A {@code flatMap} function may also be used to convert a warning into an
+ * {@code onError} signal:
+ * </p><pre>{@code
+ * result.flatMap(segment -> {
+ *   if (segment instanceof OracleR2dbcWarning) {
+ *     return errorPublisher(((OracleR2dbcWarning)segment).warning());
+ *   }
+ *   else {
+ *     ... handle other segment types ...
+ *   }
+ * })
+ * }</pre>
+ * @since 1.1.0
+ */
+public interface OracleR2dbcWarning extends Result.Message {
+
+}

src/main/java/oracle/r2dbc/impl/OracleR2dbcExceptions.java

@@ -163,12 +163,26 @@ static void requireOpenConnection(java.sql.Connection jdbcConnection) {
    * as the specified {@code sqlException}. Not null.
    */
   static R2dbcException toR2dbcException(SQLException sqlException) {
+    return toR2dbcException(sqlException, getSql(sqlException));
+  }
+
+  /**
+   * Converts a {@link SQLException} into an {@link R2dbcException}, as
+   * specified by {@link #toR2dbcException(SQLException)}. This method accepts
+   * a SQL string argument. It should be used in cases where the SQL can not
+   * be extracted by {@link #getSql(SQLException)}.
+   * @param sqlException A {@code SQLException} to convert. Not null.
+   * @param sql SQL that caused the exception
+   * @return an {@code R2dbcException} that indicates the same error conditions
+   * as the specified {@code sqlException}. Not null.
+   */
+  static R2dbcException toR2dbcException(
+    SQLException sqlException, String sql) {
     assert sqlException != null : "sqlException is null";
 
     final String message = sqlException.getMessage();
     final String sqlState = sqlException.getSQLState();
     final int errorCode = sqlException.getErrorCode();
-    final String sql = getSql(sqlException);
 
     if (sqlException instanceof SQLNonTransientException) {
       if (sqlException instanceof SQLSyntaxErrorException) {

src/main/java/oracle/r2dbc/impl/OracleResultImpl.java

@@ -27,6 +27,7 @@
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
+import oracle.r2dbc.OracleR2dbcWarning;
 import oracle.r2dbc.impl.ReadablesMetadata.RowMetadataImpl;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -150,6 +151,8 @@ private <T extends Segment, U> Publisher<U> publishSegments(
       Flux.from(publishSegments(segment -> {
         if (type.isInstance(segment))
           return mappingFunction.apply(type.cast(segment));
+        else if (segment instanceof OracleR2dbcWarning)
+          return (U)FILTERED;
         else if (segment instanceof Message)
           throw ((Message)segment).exception();
         else
@@ -398,14 +401,15 @@ static OracleResultImpl createErrorResult(R2dbcException r2dbcException) {
    * Creates a {@code Result} that publishes a {@code warning} as a
    * {@link Message} segment, followed by any {@code Segment}s of a
    * {@code result}.
+   * @param sql The SQL that resulted in a waring. Not null.
    * @param warning Warning to publish. Not null.
    * @param result Result to publisher. Not null.
    * @return A {@code Result} for a {@code Statement} execution that
    * completed with a warning.
    */
   static OracleResultImpl createWarningResult(
-    SQLWarning warning, OracleResultImpl result) {
-    return new WarningResult(warning, result);
+    String sql, SQLWarning warning, OracleResultImpl result) {
+    return new WarningResult(sql, warning, result);
   }
 
   /**
@@ -627,6 +631,9 @@ <T> Publisher<T> publishSegments(Function<Segment, T> mappingFunction) {
    */
   private static final class WarningResult extends OracleResultImpl {
 
+    /** The SQL that resulted in a warning */
+    private final String sql;
+
     /** The warning of this result */
     private final SQLWarning warning;
 
@@ -636,11 +643,13 @@ private static final class WarningResult extends OracleResultImpl {
     /**
      * Constructs a result that publishes a {@code warning} as a
      * {@link Message}, and then publishes the segments of a {@code result}.
+     * @param sql The SQL that resulted in a warning
      * @param warning Warning to publish. Not null.
      * @param result Result of segments to publish after the warning. Not null.
      */
     private WarningResult(
-      SQLWarning warning, OracleResultImpl result) {
+      String sql, SQLWarning warning, OracleResultImpl result) {
+      this.sql = sql;
       this.warning = warning;
       this.result = result;
     }
@@ -649,8 +658,11 @@ private WarningResult(
     <T> Publisher<T> publishSegments(Function<Segment, T> mappingFunction) {
       return Flux.fromStream(Stream.iterate(
         warning, Objects::nonNull, SQLWarning::getNextWarning)
-        .map(OracleR2dbcExceptions::toR2dbcException)
-        .map(MessageImpl::new))
+        .map(nextWarning ->
+          // It is noted that SQL can not be extracted from Oracle JDBC's
+          // SQLWarning objects, so it must be explicitly provided here.
+          OracleR2dbcExceptions.toR2dbcException(warning, sql))
+        .map(WarningImpl::new))
         .map(mappingFunction)
         // Invoke publishSegments(Class, Function) rather than
         // publishSegments(Function) to update the state of the result; Namely,
@@ -774,7 +786,7 @@ public long value() {
   /**
    * Implementation of {@link Message}.
    */
-  private static final class MessageImpl implements Message {
+  private static class MessageImpl implements Message {
 
     private final R2dbcException exception;
 
@@ -801,6 +813,24 @@ public String sqlState() {
     public String message() {
       return exception.getMessage();
     }
+
+    @Override
+    public String toString() {
+      return exception.toString();
+    }
+  }
+
+  /**
+   * Implementation of {@link OracleR2dbcWarning}.
+   */
+  private static final class WarningImpl
+    extends MessageImpl
+    implements OracleR2dbcWarning {
+
+    private WarningImpl(R2dbcException exception) {
+      super(exception);
+    }
+
   }
 
   /**

src/main/java/oracle/r2dbc/impl/OracleStatementImpl.java

@@ -1132,7 +1132,7 @@ private OracleResultImpl getWarnings(OracleResultImpl result) {
         preparedStatement.clearWarnings();
         return warning == null
           ? result
-          : OracleResultImpl.createWarningResult(warning, result);
+          : OracleResultImpl.createWarningResult(sql, warning, result);
       });
     }