001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.juneau.beanquery.sql;
018
019import static org.apache.juneau.commons.utils.Shorts.*;
020
021import java.sql.*;
022import java.time.*;
023import java.util.*;
024import java.util.function.*;
025import java.util.logging.*;
026import java.util.stream.*;
027
028import org.apache.juneau.commons.beanquery.*;
029import org.apache.juneau.commons.collections.*;
030
031/**
032 * A {@link BeanQuerySession} on one JDBC connection.
033 *
034 * <p>
035 * The connection is either passed in with {@link Builder#connection(Connection)} (the caller owns it and this session
036 * never closes it) or taken from the context's connection source (this session gives it back on {@link #close()}).
037 * {@link #stream(BeanQuery)} and {@link #streamValues(BeanQuery)} read rows on demand; closing the stream, or the
038 * session, closes the cursor.  Every statement gets the context's fetch size and query timeout, and, for a
039 * statement whose search includes {@code $regex}, a bounded default timeout if no explicit {@code queryTimeout} was
040 * configured.  A closed session rejects further queries with {@link IllegalStateException}.
041 *
042 * <p>
043 * The counts from {@link #find(BeanQuery)} and {@link #count(BeanQuery)} come from separate statements.  They are a
044 * consistent snapshot only if the caller's connection runs them in one suitably isolated transaction.
045 *
046 * <p>
047 * A failing statement becomes a {@link BeanQueryExecutionException} naming only the table; the SQL text (with
048 * {@code ?} placeholders) is logged at {@link Level#WARNING} with the driver's exception, whose message may include
049 * values.
050 *
051 * <h5 class='section'>Example:</h5>
052 * <p class='bjava'>
053 *    <jc>// Caller-owned connection and transaction; a per-request tenant guard.</jc>
054 *    <jk>try</jk> (SqlBeanQuerySession&lt;Person&gt; <jv>session</jv> = <jv>context</jv>.createSession()
055 *          .connection(<jv>connection</jv>)
056 *          .guard(SqlFragment.<jsm>of</jsm>(<js>"\"tenant\" = ?"</js>, <jv>tenant</jv>))
057 *          .maxLimit(200)
058 *          .build()) {
059 *       Page&lt;Person&gt; <jv>page</jv> = <jv>session</jv>.find(<jv>query</jv>);
060 *       <jk>try</jk> (Stream&lt;Map&lt;String,Object&gt;&gt; <jv>rows</jv> = <jv>session</jv>.streamValues(<jv>exportQuery</jv>)) {
061 *          <jv>rows</jv>.forEach(<jv>csv</jv>::write);
062 *       }
063 *    }
064 * </p>
065 *
066 * @param <T> The row (bean) type.
067 * @since 10.0.0
068 */
069public final class SqlBeanQuerySession<T> extends BeanQuerySession<T> {
070
071   private static final Logger LOG = Logger.getLogger(SqlBeanQuerySession.class.getName());
072
073   // Ceiling for the timeout seconds handed to PreparedStatement.setQueryTimeout(int).  Clamped below
074   // Integer.MAX_VALUE, not at it: at least one JDBC driver (H2 2.3.232) computes seconds*1000 as a 32-bit int
075   // inside its own setQueryTimeout(int), which overflows near Integer.MAX_VALUE seconds -- this bound keeps that
076   // downstream multiplication safe too, while still being functionally "no timeout" for any real use (~24.8 days).
077   private static final int MAX_QUERY_TIMEOUT_SECONDS = Integer.MAX_VALUE / 1000;
078
079   /**
080    * Builder for {@link SqlBeanQuerySession}.  See the class Javadoc for an example.
081    *
082    * @param <T> The row type.
083    */
084   public static final class Builder<T> extends BeanQuerySession.Builder<T,Builder<T>> {
085
086      private final SqlBeanQueryContext<T> context;
087      private Connection connection;
088      private SqlFragment guard;
089
090      Builder(SqlBeanQueryContext<T> context) {
091         super(context);
092         this.context = context;
093      }
094
095      /**
096       * Runs this session on a caller-owned connection instead of one from the context's connection source.  The
097       * session never closes it.
098       *
099       * @param value The connection.  Must not be <jk>null</jk>.
100       * @return This object.
101       */
102      public Builder<T> connection(Connection value) {
103         connection = reqnn("value", value);
104         return this;
105      }
106
107      /**
108       * Sets a per-request SQL predicate, AND-ed (in parentheses) with the context guard into every statement.
109       *
110       * @param value The guard, or <jk>null</jk> for none.
111       * @return This object.
112       */
113      public Builder<T> guard(SqlFragment value) {
114         guard = value;
115         return this;
116      }
117
118      /**
119       * {@inheritDoc}
120       *
121       * @throws BeanQueryExecutionException If no connection was given and none can be obtained.
122       */
123      @Override /* BeanQuerySession.Builder */
124      public SqlBeanQuerySession<T> build() {
125         return new SqlBeanQuerySession<>(this);
126      }
127   }
128
129   private final SqlBeanQueryContext<T> context;
130   private final SqlFragment guard;
131   private final Connection connection;
132   private final boolean owned;
133   private final List<Cursor<?>> open = new ArrayList<>();
134   private boolean closed;
135
136   SqlBeanQuerySession(Builder<T> builder) {
137      super(builder);  // Validates the settings before a connection is taken.
138      context = builder.context;
139      guard = builder.guard;
140      owned = builder.connection == null;
141      connection = owned ? context.acquire() : builder.connection;
142   }
143
144   /**
145    * {@inheritDoc}
146    *
147    * @return The context that created this session.
148    */
149   @Override /* BeanQuerySession */
150   public SqlBeanQueryContext<T> getContext() {
151      return context;
152   }
153
154   @Override /* BeanQuerySession */
155   protected Page<T> doFind(ResolvedQuery query) {
156      var mapper = context.rowMapper();
157      var stmt = context.renderRows(query, guard);
158      return page(list(stmt, rs -> mapper.map(rs, stmt.columns())), query, () -> countRows(query, false), () -> countRows(query, true));
159   }
160
161   @Override /* BeanQuerySession */
162   protected Page<Map<String,Object>> doFindValues(ResolvedQuery query) {
163      var stmt = context.renderRows(query, guard);
164      return page(list(stmt, rs -> valueRow(rs, stmt.columns())), query, () -> countRows(query, false), () -> countRows(query, true));
165   }
166
167   @Override /* BeanQuerySession */
168   protected Counts doCount(ResolvedQuery query) {
169      return Counts.of(countRows(query, false), countRows(query, true));
170   }
171
172   @Override /* BeanQuerySession */
173   protected Stream<T> doStream(ResolvedQuery query) {
174      var mapper = context.rowMapper();
175      var stmt = context.renderRows(query, guard);
176      return stream(new Cursor<>(stmt, rs -> mapper.map(rs, stmt.columns())));
177   }
178
179   @Override /* BeanQuerySession */
180   protected Stream<Map<String,Object>> doStreamValues(ResolvedQuery query) {
181      var stmt = context.renderRows(query, guard);
182      return stream(new Cursor<>(stmt, rs -> valueRow(rs, stmt.columns())));
183   }
184
185   /**
186    * Closes any open streams, then gives back the connection if this session took it from the context.  Calling it
187    * again does nothing.  After this, every query method throws {@link IllegalStateException}.
188    *
189    * @throws BeanQueryExecutionException If the connection cannot be released.
190    */
191   @Override /* BeanQuerySession */
192   public void close() {
193      if (closed)
194         return;
195      closed = true;
196      for (var c : new ArrayList<>(open))
197         c.closeBySession();
198      if (owned)
199         context.release(connection);
200   }
201
202   private long countRows(ResolvedQuery query, boolean applySearch) {
203      var out = list(context.renderCount(query, applySearch, guard), rs -> rs.getLong(1));
204      return out.isEmpty() ? 0L : out.get(0);
205   }
206
207   private <R> List<R> list(SqlStatement stmt, RowReader<R> reader) {
208      assertOpen();  // Outside the try: a closed-session IllegalStateException must surface unwrapped.
209      try (var ps = prepare(stmt); var rs = ps.executeQuery()) {
210         var out = new ArrayList<R>();
211         while (rs.next())
212            out.add(reader.read(rs));
213         return out;
214      } catch (BeanQueryExecutionException | BeanQuerySyntaxException e) {
215         throw e;
216      } catch (SQLException | RuntimeException e) {  // Includes the rowMapper throwing.
217         throw failed(stmt, e);
218      }
219   }
220
221   private void assertOpen() {
222      if (closed)
223         throw isex("SqlBeanQuerySession is closed.");
224   }
225
226   private PreparedStatement prepare(SqlStatement stmt) throws SQLException {
227      assertOpen();
228      var ps = connection.prepareStatement(stmt.sql());
229      try {
230         ps.setFetchSize(context.fetchSize);
231         var timeout = effectiveTimeout(stmt);
232         if (timeout != null)
233            ps.setQueryTimeout((int)Math.min(MAX_QUERY_TIMEOUT_SECONDS, Math.max(1, queryTimeoutSeconds(timeout))));
234         var binds = stmt.binds();
235         for (var i = 0; i < binds.size(); i++)
236            ps.setObject(i + 1, binds.get(i));
237         return ps;
238      } catch (SQLException | RuntimeException e) {
239         closeQuietly(ps);
240         throw e;
241      }
242   }
243
244   private Duration effectiveTimeout(SqlStatement stmt) {
245      if (context.queryTimeout != null)
246         return context.queryTimeout;
247      return stmt.hasRegex() ? SqlBeanQueryContext.DEFAULT_REGEX_TIMEOUT : null;
248   }
249
250   // Rounds up to the next whole second without ever computing total milliseconds: Duration.toMillis() can throw
251   // ArithmeticException for a Duration whose millis would overflow long; getSeconds()/getNano() cannot.
252   // The "seconds < Long.MAX_VALUE" guard exists specifically so the "+ 1" itself cannot overflow when
253   // getSeconds() is already at Long.MAX_VALUE (reachable via the public builder, e.g.
254   // Duration.ofSeconds(Long.MAX_VALUE, 999_999_999)) -- without it, Long.MAX_VALUE + 1 wraps to Long.MIN_VALUE,
255   // which would silently survive the caller's Math.max(1, ...) clamp as 1, producing a 1-second timeout instead
256   // of the intended effectively-no-timeout behavior.
257   private static long queryTimeoutSeconds(Duration d) {
258      var seconds = d.getSeconds();
259      return (d.getNano() > 0 && seconds < Long.MAX_VALUE) ? seconds + 1 : seconds;
260   }
261
262   private BeanQueryExecutionException failed(SqlStatement stmt, Throwable e) {
263      LOG.log(Level.WARNING, e, () -> "Bean query failed: " + stmt.sql());  // SQL text with placeholders; the driver's message may include values.
264      return new BeanQueryExecutionException(e, "Query execution failed for table '%s'.", context.table);
265   }
266
267   private <R> Stream<R> stream(Cursor<R> cursor) {  // Not static: Cursor is an inner class.
268      return StreamSupport.stream(cursor, false).onClose(cursor::close);
269   }
270
271   private static Map<String,Object> valueRow(ResultSet rs, List<String> columns) throws SQLException {
272      var keys = columns.toArray(new String[0]);
273      var vals = new Object[keys.length];
274      for (var i = 0; i < keys.length; i++)
275         vals[i] = rs.getObject(i + 1);
276      return new SimpleMap<>(keys, vals);
277   }
278
279   private static void closeQuietly(AutoCloseable c) {
280      try {
281         c.close();
282      } catch (Exception e) {  // NOSONAR - A close failure must not hide the real result or error.
283         LOG.log(Level.FINE, "Failed to close a JDBC resource.", e);
284      }
285   }
286
287   @FunctionalInterface
288   private interface RowReader<R> {
289      R read(ResultSet rs) throws SQLException;
290   }
291
292   /** An open statement and result set, read one row per {@link #tryAdvance(Consumer)}. */
293   private final class Cursor<R> extends Spliterators.AbstractSpliterator<R> implements AutoCloseable {
294
295      private final SqlStatement stmt;
296      private final RowReader<R> reader;
297      private PreparedStatement ps;
298      private ResultSet rs;
299      private boolean done;
300      private boolean closedBySession;  // Set only by closeBySession() -- distinguishes a forced shutdown from EOF.
301
302      Cursor(SqlStatement stmt, RowReader<R> reader) {
303         super(Long.MAX_VALUE, ORDERED);
304         this.stmt = stmt;
305         this.reader = reader;
306         assertOpen();  // Outside the try: a closed-session IllegalStateException must surface unwrapped.
307         try {
308            ps = prepare(stmt);
309            rs = ps.executeQuery();
310         } catch (BeanQueryExecutionException | BeanQuerySyntaxException e) {
311            close();
312            throw e;
313         } catch (SQLException | RuntimeException e) {
314            close();
315            throw failed(stmt, e);
316         }
317         open.add(this);
318      }
319
320      @Override /* Spliterator */
321      public boolean tryAdvance(Consumer<? super R> action) {
322         if (done) {
323            if (closedBySession)
324               throw isex("SqlBeanQuerySession was closed while a stream from it was still open.");
325            return false;
326         }
327         R row;
328         try {
329            if (! rs.next()) {
330               close();
331               return false;
332            }
333            row = reader.read(rs);
334         } catch (BeanQueryExecutionException | BeanQuerySyntaxException e) {
335            close();
336            throw e;
337         } catch (SQLException | RuntimeException e) {  // Includes the rowMapper throwing.
338            close();
339            throw failed(stmt, e);
340         }
341         // action.accept() runs outside the try: it's the caller's stream-pipeline code (e.g. a downstream
342         // Collectors.toMap duplicate-key check), not our SQL/row-mapping logic, so its exceptions must propagate
343         // unwrapped rather than being misreported as a query failure.
344         action.accept(row);
345         return true;
346      }
347
348      @Override /* AutoCloseable */
349      public void close() {
350         if (done)
351            return;
352         done = true;
353         open.remove(this);
354         if (rs != null)
355            closeQuietly(rs);
356         if (ps != null)
357            closeQuietly(ps);
358      }
359
360      /**
361       * Closes this cursor because the owning session is closing, not because the caller is done with it. A later
362       * {@link #tryAdvance} throws {@link IllegalStateException} instead of looking like the stream simply ran out
363       *. Called only from {@code SqlBeanQuerySession.close()}'s force-close loop over still-open cursors.
364       */
365      void closeBySession() {
366         closedBySession = true;
367         close();
368      }
369   }
370}