import org.jooq.Cursor; //导入依赖的package包/类
private Select(
  Connection connection,
  QueryBuilder<ResultQuery<? extends R>> queryBuilder,
  RecordMapper<? super R, ? extends T> recordMapper
) {
  super(
    subscriber -> {
      try (ResultQuery<? extends R> query = queryBuilder.build(connection)) {
        Cursor<? extends R> cursor = query.fetchLazy();
        setupUnsubscription(subscriber, query, cursor);
        log.debug("Select setProducer for  {}", query);
        subscriber.setProducer(new SelectProducer<>(
          subscriber,
          query,
          cursor,
          recordMapper
        ));
      } catch (Throwable t) {
        handleException(t, subscriber);
      }
    }
  );
}
 

import org.jooq.Cursor; //导入依赖的package包/类
protected Iterator<AllocationCandidate> iteratorHosts(List<Long> volumes, QueryOptions options, boolean hosts,
        AllocationCandidateCallback callback) {
    final Cursor<Record2<Long,Long>> cursor = create()
            .select(HOST.ID, STORAGE_POOL.ID)
            .from(HOST)
            .leftOuterJoin(STORAGE_POOL_HOST_MAP)
                .on(STORAGE_POOL_HOST_MAP.HOST_ID.eq(HOST.ID)
                    .and(STORAGE_POOL_HOST_MAP.REMOVED.isNull()))
            .join(STORAGE_POOL)
                .on(STORAGE_POOL.ID.eq(STORAGE_POOL_HOST_MAP.STORAGE_POOL_ID))
            .leftOuterJoin(AGENT)
                .on(AGENT.ID.eq(HOST.AGENT_ID))
            .where(
                AGENT.ID.isNull().or(AGENT.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(HOST.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(STORAGE_POOL.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(getQueryOptionCondition(options)))
            .orderBy(SPREAD.get() ? HOST.COMPUTE_FREE.desc() : HOST.COMPUTE_FREE.asc())
            .fetchLazy();

    return new AllocationCandidateIterator(objectManager, cursor, volumes, hosts, callback);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
protected Iterator<AllocationCandidate> iteratorHosts(List<Long> volumes, QueryOptions options, boolean hosts) {
    final Cursor<Record2<Long,Long>> cursor = create()
            .select(HOST.ID, STORAGE_POOL.ID)
            .from(HOST)
            .leftOuterJoin(STORAGE_POOL_HOST_MAP)
                .on(STORAGE_POOL_HOST_MAP.HOST_ID.eq(HOST.ID)
                    .and(STORAGE_POOL_HOST_MAP.REMOVED.isNull()))
            .join(STORAGE_POOL)
                .on(STORAGE_POOL.ID.eq(STORAGE_POOL_HOST_MAP.STORAGE_POOL_ID))
            .leftOuterJoin(AGENT)
                .on(AGENT.ID.eq(HOST.AGENT_ID))
            .where(
                AGENT.ID.isNull().or(AGENT.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(HOST.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(STORAGE_POOL.STATE.eq(CommonStatesConstants.ACTIVE))
                .and(getQueryOptionCondition(options)))
            .orderBy(SPREAD.get() ? HOST.COMPUTE_FREE.asc() : HOST.COMPUTE_FREE.desc())
            .fetchLazy();

    return new AllocationCandidateIterator(objectManager, cursor, volumes, hosts);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private List<String> fetchList (ResultQuery<Record1<String>> query, String columnField) {
  Cursor<Record1<String>> cursor = null;
  List<String> result = new ArrayList<String> (1000);
  try {
    cursor = query.fetchLazy ();
    // Cursor has similar methods as Iterator<R>
    while (cursor.hasNext ()) {
      Record record = cursor.fetchOne ();
      String sValue = (String) record.getValue (columnField);
      result.add (sValue);
    }
  } finally {
    if (cursor != null) {
      cursor.close ();
    }
  }
  return result;
}
 

import org.jooq.Cursor; //导入依赖的package包/类
List<DependencyLink> aggregateDependencies(long endTs, @Nullable Long lookback, Connection conn) {
  endTs = endTs * 1000;
  // Lazy fetching the cursor prevents us from buffering the whole dataset in memory.
  Cursor<Record> cursor = context.get(conn)
      .selectDistinct(hasTraceIdHigh.get() ? LINK_FIELDS : LINK_FIELDS_WITHOUT_TRACE_ID_HIGH)
      // left joining allows us to keep a mapping of all span ids, not just ones that have
      // special annotations. We need all span ids to reconstruct the trace tree. We need
      // the whole trace tree so that we can accurately skip local spans.
      .from(ZIPKIN_SPANS.leftJoin(ZIPKIN_ANNOTATIONS)
          // NOTE: we are intentionally grouping only on the low-bits of trace id. This buys time
          // for applications to upgrade to 128-bit instrumentation.
          .on(ZIPKIN_SPANS.TRACE_ID.eq(ZIPKIN_ANNOTATIONS.TRACE_ID).and(
              ZIPKIN_SPANS.ID.eq(ZIPKIN_ANNOTATIONS.SPAN_ID)))
          .and(ZIPKIN_ANNOTATIONS.A_KEY.in(CLIENT_SEND, CLIENT_ADDR, SERVER_RECV, SERVER_ADDR)))
      .where(lookback == null ?
          ZIPKIN_SPANS.START_TS.lessOrEqual(endTs) :
          ZIPKIN_SPANS.START_TS.between(endTs - lookback * 1000, endTs))
      // Grouping so that later code knows when a span or trace is finished.
      .groupBy(hasTraceIdHigh.get()
          ? LINK_GROUP_FIELDS
          : LINK_GROUP_FIELDS_WITHOUT_TRACE_ID_HIGH).fetchLazy();

  Iterator<Iterator<DependencyLinkSpan>> traces =
      new DependencyLinkSpanIterator.ByTraceId(cursor.iterator(), hasTraceIdHigh.get());

  if (!traces.hasNext()) return Collections.emptyList();

  DependencyLinker linker = new DependencyLinker();

  while (traces.hasNext()) {
    linker.putTrace(traces.next());
  }

  return linker.link();
}
 

import org.jooq.Cursor; //导入依赖的package包/类
/**
 * Close the cursor quietly
 */
public static <R extends Record> void closeQuietly(Cursor<R> cursor) {
  try {
    if (cursor != null && !cursor.isClosed()) {
      cursor.close();
      log.debug("closed {}", cursor);
    }
  } catch (Exception e) {
    log.debug(e.getMessage(), e);
  }
}
 

import org.jooq.Cursor; //导入依赖的package包/类
public SelectProducer(
  Subscriber<? super T> subscriber,
  ResultQuery<? extends R> query,
  Cursor<? extends R> cursor,
  RecordMapper<? super R, ? extends T> recordMapper
) {
  this.subscriber = subscriber;
  this.query = query;
  this.cursor = cursor;
  this.recordMapper = recordMapper;
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private static <R extends Record, T> void setupUnsubscription(
  Subscriber<? super T> subscriber,
  ResultQuery<? extends R> query,
  Cursor<? extends R> cursor
) {
  subscriber.add(
    Subscriptions.create(
      () -> {
        closeQuietly(cursor);
        closeQuietly(query);
      }
    )
  );
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Transactional(readOnly = true)
public Collection<AuthorWithBooks> findAuthorsWithBooksJooqFetchLazyOldFashionGroupBy() {

	try (Cursor<Record> records = dslContext.select()
			.from(AUTHOR.leftOuterJoin(BOOK).on(BOOK.AUTHOR_ID.equal(AUTHOR.ID)))
			.fetchLazy()) {

		Map<Long, AuthorWithBooks> booksMap = new HashMap<>();

		for (Record r : records) {

			Long authorId = r.getValue(TAuthor.AUTHOR.ID);

			AuthorWithBooks authorWithBooks = booksMap.get(authorId);

			if (authorWithBooks == null) {
				authorWithBooks = new AuthorWithBooks();
				authorWithBooks.setAuthor(new Author(authorId, r.getValue(TAuthor.AUTHOR.NAME)));
				authorWithBooks.setBooks(new ArrayList<>());
				booksMap.put(authorId, authorWithBooks);
			}

			Book book = new Book(r.getValue(TBook.BOOK.ID), r.getValue(TBook.BOOK.TITLE), authorId);
			authorWithBooks.getBooks().add(book);
		}

		return booksMap.values();
	}
}
 

import org.jooq.Cursor; //导入依赖的package包/类
public AllocationCandidateIterator(ObjectManager objectManager, Cursor<Record2<Long,Long>> cursor, List<Long> volumeIds, boolean hosts,
        AllocationCandidateCallback callback) {
    super();
    this.objectManager = objectManager;
    this.volumeIds = volumeIds;
    this.cursor = cursor;
    this.hosts = hosts;
    this.callback = callback;
}
 

import org.jooq.Cursor; //导入依赖的package包/类
public AllocationCandidateIterator(ObjectManager objectManager, Cursor<Record2<Long,Long>> cursor, List<Long> volumeIds, boolean hosts) {
    super();
    this.objectManager = objectManager;
    this.volumeIds = volumeIds;
    this.cursor = cursor;
    this.hosts = hosts;
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Override
public Iterable<Tag> get(DaoFilter filter) throws DaoException {
    Collection<Condition> conditions = new ArrayList<Condition>();
    if (filter.getTags() != null) {
        conditions.add(Tables.TAGS.RAW_TAG.in(filter.getTags()));
    }
    if (filter.getConcepts() != null) {
        conditions.add(Tables.TAGS.CONCEPT_ID.in(filter.getConceptIds()));
    }
    Cursor<Record> cursor = fetchLazy(conditions);
    return buildTagIterable(cursor);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private Iterable<Tag> buildTagIterable(Cursor<Record> cursor) {
    return new SqlDaoIterable<Tag>(cursor) {
        @Override
        public Tag transform(Record record) throws DaoException {
            return buildTag(record);
        }
    };
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Override
public Iterable<Concept> get(DaoFilter filter) throws DaoException {
    Collection<Condition> conditions = new ArrayList<Condition>();
    if (filter.getConceptIds() != null) {
        conditions.add(Tables.CONCEPTS.CONCEPT_ID.in(filter.getConceptIds()));
    }
    Cursor<Record> cursor = fetchLazy(conditions);
    return buildConceptIterable(cursor);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private Iterable<Concept> buildConceptIterable(Cursor<Record> cursor) {
    return new SqlDaoIterable<Concept>(cursor) {
        @Override
        public Concept transform(Record record) throws DaoException {
            return buildConcept(record);
        }
    };
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Override
public Iterable<User> get(DaoFilter filter) throws DaoException {
    Collection<Condition> conditions = new ArrayList<Condition>();
    if (filter.getUserIds() != null) {
        conditions.add(Tables.USERS.USER_ID.in(filter.getUserIds()));
    }
    Cursor<Record> cursor = fetchLazy(conditions);
    return buildUserIterable(cursor);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private Iterable<User> buildUserIterable(Cursor<Record> cursor) {
    return new SqlDaoIterable<User>(cursor) {
        @Override
        public User transform(Record record) throws DaoException {
            return buildUser(record);
        }
    };
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Override
public Iterable<Item> get(DaoFilter filter) throws DaoException {
    Collection<Condition> conditions = new ArrayList<Condition>();
    if (filter.getUserIds() != null) {
        conditions.add(Tables.ITEMS.ITEM_ID.in(filter.getUserIds()));
    }
    Cursor<Record> cursor = fetchLazy(conditions);
    return buildItemIterable(cursor);
}
 

import org.jooq.Cursor; //导入依赖的package包/类
private Iterable<Item> buildItemIterable(Cursor<Record> cursor) {
    return new SqlDaoIterable<Item>(cursor) {
        @Override
        public Item transform(Record record) throws DaoException {
            return buildItem(record);
        }
    };
}
 

import org.jooq.Cursor; //导入依赖的package包/类
@Test
@Ignore @SuppressWarnings("unused")
public void testSimpleSelectAll () {
  Table<Record> table = tableByName (tsvFileName);

  long startTime = System.nanoTime ();
  int count = 0;
  ResultQuery<Record> query = context.selectFrom (table);
  // log.debug ("PresetValuesFlatTable sql:"+query.getSQL ());
  Cursor<Record> cursor = null;
  try {
    cursor = query.fetchLazy ();
    // Cursor has similar methods as Iterator<R>
    while (cursor.hasNext ()) {
      Record record = cursor.fetchOne ();
      for (Field<?> columnField : record.fields ()) {
        String value = (String) record.getValue (columnField);
        count++;
      }
    }
  } finally {
    if (cursor != null) {
      cursor.close ();
    }
  }

  long endTime = System.nanoTime ();
  long duration = endTime - startTime;
  log.debug ("flat-count:" + count + ", duration: " + duration);
}