{@inheritDoc}

@see jp.co.future.uroborosql.fluent.SqlQuery#stream(jp.co.future.uroborosql.converter.ResultSetConverter)
961527
@Override
	public <T> Stream<T> stream(final ResultSetConverter<T> converter) {
		try {
			return agent().query(context(), converter);
		} catch (SQLException e) {
			throw new UroborosqlSQLException(e);
		}
	}
608208
@Override
    public InputStream getBinaryStream(int columnIndex) throws SQLException {
        checkColumnBounds(columnIndex);
        try {
            return new ByteArrayInputStream(
                    table.getStringAsBytes(columnIndex - 1));
        } catch (Exception x) {
            throw SQLError.get(x);
        }
    }
203690
public static Object convertBlob(Object blob, InputStream input) throws SQLException {
        return convertBlob(blob, toByteArray(input));
    }
315711
@Override
	public SqlBatch paramStream(final Stream<?> stream) {
		if (this.stream == null) {
			this.stream = stream;
		} else {
			this.stream = Stream.concat(this.stream, stream);
		}
		return this;
	}
812494
@Override
	public <E> Stream<E> doSelect(final SqlAgent agent, final SqlContext context, final Class<? extends E> entityType)
			throws SQLException {
		return agent.query(context, new EntityResultSetConverter<>(entityType, new PropertyMapperManager(
				propertyMapperManager)));
	}
676444
@Override
	public void updateNCharacterStream(final int columnIndex, final Reader x) throws SQLException {
		wrapped.updateNCharacterStream(columnIndex, x);
	}
915724
@Override
    public <T> Flowable<Tx<T>> get(@Nonnull ResultSetMapper<? extends T> mapper) {
        Preconditions.checkNotNull(mapper, "mapper cannot be null");
        return Flowable.defer(() -> {
            AtomicReference<Connection> connection = new AtomicReference<Connection>();
            Flowable<T> o = Update.<T>createReturnGeneratedKeys( //
                    update.updateBuilder.connections //
                            .map(c -> Util.toTransactedConnection(connection, c)),
                    update.parameterGroupsToFlowable(), update.updateBuilder.sql, mapper, false);
            return o.materialize() //
                    .flatMap(n -> Tx.toTx(n, connection.get(), db)) //
                    .doOnNext(tx -> {
                        if (tx.isComplete()) {
                            ((TxImpl<T>) tx).connection().commit();
                        }
                    });
        });
    }
790105
@Local
    @Override
    public <V> V directQuery(String sql, Function<ResultSet, V> handler) {
        final Connection conn = readPool.poll();
        try {
            if (logger.isLoggable(Level.FINEST)) logger.finest("direct query sql=" + sql);
            conn.setReadOnly(true);
            final Statement statement = conn.createStatement();
            //final PreparedStatement statement = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            final ResultSet set = statement.executeQuery(sql);// ps.executeQuery();
            V rs = handler.apply(set);
            set.close();
            statement.close();
            return rs;
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        } finally {
            if (conn != null) readPool.offerConnection(conn);
        }
    }
320442
public StreamingOutput stream(Context context,
                              String pid,
                              String format,
                              String exportContext,
                              String encoding) throws ServerException {
        return delegate.stream(context, pid, format, exportContext, encoding);
    }
1072521
public final static <T> CompletableFuture<List<T>> streamToCompletableFuture(final Stream<T> stream) {
        return CompletableFuture.completedFuture(stream.collect(Collectors.toList()));

    }
912486
@Override
    public void set(BindingSetStatementContext<JsonArray> ctx) throws SQLException {
        ctx.statement().setString(ctx.index(), Objects.toString(ctx.convert(converter()).value(), null));
    }
1075120
public Object javaToSql(Object source)
    {
        if (source == null)
            return null;
        try
        {
            ByteArrayOutputStream bao = new ByteArrayOutputStream();
            GZIPOutputStream gos = new GZIPOutputStream(bao);
            ObjectOutputStream oos = new ObjectOutputStream(gos);
            oos.writeObject(source);
            oos.close();
            gos.close();
            bao.close();
            byte[] result = bao.toByteArray();
            return result;
        }
        catch (Throwable t)
        {
            throw new ConversionException(t);
        }
    }
1154426
@Override
  public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState) throws SchemaConversionException {
    init(workUnitState);
    this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(),
        Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build();
    Flowable<StreamEntity<DO>> outputStream =
        inputStream.getRecordStream()
            .flatMap(in -> {
              if (in instanceof ControlMessage) {
                ControlMessage out = (ControlMessage) in;

                getMessageHandler().handleMessage((ControlMessage) in);

                // update the output schema with the new input schema from the MetadataUpdateControlMessage
                if (in instanceof MetadataUpdateControlMessage) {
                  this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(
                      ((MetadataUpdateControlMessage) in).getGlobalMetadata(),
                      Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata()
                          .getSchema(), workUnitState))).build();
                  out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata);
                }

                return Flowable.just(((ControlMessage<DO>) out));
              } else if (in instanceof RecordEnvelope) {
                RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in;
                return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState);
              } else {
                throw new UnsupportedOperationException();
              }
            }, 1);
    outputStream = outputStream.doOnComplete(this::close);
    return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata);
  }
811457
public static String fromResultSet(ResultSet resultSet) throws SQLException {
    if (resultSet == null) throw new NullPointerException("resultSet == null");
    if (!resultSet.isBeforeFirst()) throw new IllegalStateException("Result set not at first.");

    List<String> headers = new ArrayList<>();
    ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
    int columnCount = resultSetMetaData.getColumnCount();
    for (int column = 0; column < columnCount; column++) {
      headers.add(resultSetMetaData.getColumnName(column + 1));
    }

    List<String[]> data = new ArrayList<>();
    while (resultSet.next()) {
      String[] rowData = new String[columnCount];
      for (int column = 0; column < columnCount; column++) {
        rowData[column] = resultSet.getString(column + 1);
      }
      data.add(rowData);
    }

    String[] headerArray = headers.toArray(new String[headers.size()]);
    String[][] dataArray = data.toArray(new String[data.size()][]);
    return FlipTable.of(headerArray, dataArray);
  }
683743
@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public T convert(ResponseBody value) throws IOException {
		try {
			return (T) binderContext.parseCollection(value.byteStream(), (Collection) clazz.newInstance(), beanClazz);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		} finally {
			value.close();
		}
	}
81881
@Override
  public ResultSet executeQuery(String sql) throws SQLException
  {
    raiseSQLExceptionIfStatementIsClosed();
    return executeQueryInternal(sql, null);
  }
475967
public static <T> Stream<T> stream(final Iterable<T> source) {
        return new Stream<T>() {
            @Override
            public Iterator<T> iterator() {
                return source.iterator();
            }
        };
    }