Page research without exact counts and bound candidate scans [skip ci]
This commit is contained in:
+170
-35
@@ -173,8 +173,31 @@ func decodeAuthors(raw string) ([]string, error) {
|
||||
return names, nil
|
||||
}
|
||||
|
||||
func (r *PostgresRepository) positions(ctx context.Context, tx *sql.Tx, query string, args []any) ([]position, error) {
|
||||
rows, err := tx.QueryContext(ctx, r.query(query), args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
positions := []position{}
|
||||
for rows.Next() {
|
||||
var p position
|
||||
// Preserve the stored title key, including the published text codec.
|
||||
if err = rows.Scan(&p.ID, &p.Key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
positions = append(positions, p)
|
||||
}
|
||||
return positions, rows.Err()
|
||||
}
|
||||
|
||||
func (r *PostgresRepository) List(ctx context.Context, filters Filters) (Page, error) {
|
||||
result := Page{Items: []Paper{}, Topics: []string{}, Page: filters.Page, PageSize: filters.PageSize}
|
||||
result := Page{Items: []Paper{}, Topics: []string{}, PageSize: filters.PageSize}
|
||||
cursor, err := decodeCursor(filters)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
original := filters
|
||||
tx, err := r.begin(ctx)
|
||||
if err != nil {
|
||||
return result, err
|
||||
@@ -185,56 +208,168 @@ func (r *PostgresRepository) List(ctx context.Context, filters Filters) (Page, e
|
||||
filters.Topic = encodeText(filters.Topic)
|
||||
}
|
||||
where, args := whereClause(filters)
|
||||
if err = tx.QueryRowContext(ctx, r.query(`SELECT count(*) FROM {schema}papers p`+where), args...).Scan(&result.Total); err != nil {
|
||||
return result, err
|
||||
if cursor.ID != 0 {
|
||||
clause := " AND "
|
||||
if where == "" {
|
||||
clause = " WHERE "
|
||||
}
|
||||
op := "<"
|
||||
if cursor.Before {
|
||||
op = ">"
|
||||
}
|
||||
args = append(args, cursor.ID)
|
||||
if filters.Sort == "title" {
|
||||
if cursor.Before {
|
||||
op = "<"
|
||||
} else {
|
||||
op = ">"
|
||||
}
|
||||
args = append(args, cursor.Key)
|
||||
where += fmt.Sprintf("%s(p.title_key,p.id) %s ($%d::text COLLATE \"C\",$%d)", clause, op, len(args), len(args)-1)
|
||||
} else {
|
||||
where += fmt.Sprintf("%sp.id %s $%d", clause, op, len(args))
|
||||
}
|
||||
}
|
||||
order := "p.id DESC"
|
||||
if filters.Sort == "title" {
|
||||
order = "p.title_key,p.id"
|
||||
}
|
||||
args = append(args, filters.PageSize, (filters.Page-1)*filters.PageSize)
|
||||
// Select IDs before loading potentially TOASTed titles/abstracts. With
|
||||
// sparse categories, selecting full rows here can make the planner walk
|
||||
// hundreds of thousands of wide heap rows before finding eight matches.
|
||||
candidateQuery := `SELECT p.id FROM {schema}papers p`
|
||||
candidateOrder := order
|
||||
prefix := ""
|
||||
if where != "" {
|
||||
columns := "p.id"
|
||||
if cursor.Before {
|
||||
order = "p.id"
|
||||
if filters.Sort == "title" {
|
||||
columns += ",p.title_key"
|
||||
order = "p.title_key DESC,p.id DESC"
|
||||
}
|
||||
// Apply search before LIMIT/ORDER BY so a sparse term cannot turn
|
||||
// into a backward heap scan that recomputes vectors for millions.
|
||||
prefix = fmt.Sprintf(`matches AS MATERIALIZED (SELECT %s FROM {schema}papers p%s), `, columns, where)
|
||||
candidateQuery = "SELECT id FROM matches"
|
||||
candidateOrder = strings.ReplaceAll(order, "p.", "")
|
||||
}
|
||||
query := fmt.Sprintf(`WITH %spage_ids AS MATERIALIZED (
|
||||
%s ORDER BY %s LIMIT $%d OFFSET $%d
|
||||
) SELECT p.id,p.title,p.abstract,p.authors_json,p.first_published
|
||||
FROM page_ids JOIN {schema}papers p ON p.id=page_ids.id ORDER BY %s`, prefix, candidateQuery, candidateOrder, len(args)-1, len(args), order)
|
||||
rows, err := tx.QueryContext(ctx, r.query(query), args...)
|
||||
if err != nil {
|
||||
return result, err
|
||||
args = append(args, filters.PageSize+1)
|
||||
columns := "p.id,p.title_key"
|
||||
if filters.Sort != "title" {
|
||||
columns = "p.id, ''::text AS title_key"
|
||||
}
|
||||
for rows.Next() {
|
||||
p, err := r.scanPaper(rows)
|
||||
// Search candidates are materialized to avoid scanning millions of wide
|
||||
// rows in ID order for sparse terms. There is no count or deep OFFSET.
|
||||
query := fmt.Sprintf("SELECT %s FROM {schema}papers p%s ORDER BY %s LIMIT $%d", columns, where, order, len(args))
|
||||
if filters.Query != "" || filters.Topic != "" {
|
||||
query = fmt.Sprintf("WITH matches AS MATERIALIZED (SELECT %s FROM {schema}papers p%s) SELECT p.id,p.title_key FROM matches p ORDER BY %s LIMIT $%d", columns, where, order, len(args))
|
||||
}
|
||||
positions := []position{}
|
||||
topicOnly := filters.Sort != "title" && filters.Topic != "" && filters.Query == ""
|
||||
if topicOnly {
|
||||
// The published paper_tags foreign key guarantees these IDs exist.
|
||||
// Walk the (tag_id,paper_id) index instead of joining all matching
|
||||
// paper rows before LIMIT. DISTINCT removes duplicate source evidence.
|
||||
topicArgs := []any{filters.Topic, filters.PageSize + 1}
|
||||
predicate := ""
|
||||
if cursor.ID != 0 {
|
||||
op := "<"
|
||||
if cursor.Before {
|
||||
op = ">"
|
||||
}
|
||||
topicArgs = append(topicArgs, cursor.ID)
|
||||
predicate = " AND pt.paper_id " + op + " $3"
|
||||
}
|
||||
direction := "DESC"
|
||||
if cursor.Before {
|
||||
direction = "ASC"
|
||||
}
|
||||
query := fmt.Sprintf(`SELECT DISTINCT pt.paper_id,''::text AS title_key FROM {schema}paper_tags pt
|
||||
WHERE pt.tag_id=(SELECT id FROM {schema}tags WHERE namespace='arxiv:category' AND value=$1)%s
|
||||
ORDER BY pt.paper_id %s LIMIT $2`, predicate, direction)
|
||||
positions, err = r.positions(ctx, tx, query, topicArgs)
|
||||
if err != nil {
|
||||
_ = rows.Close()
|
||||
return result, err
|
||||
}
|
||||
result.Items = append(result.Items, p)
|
||||
}
|
||||
err = rows.Err()
|
||||
_ = rows.Close()
|
||||
if err != nil {
|
||||
return result, err
|
||||
if !topicOnly && filters.Sort != "title" && (filters.Query != "" || filters.Topic != "") {
|
||||
// Common terms can fill a page near its cursor. Probe a bounded ID
|
||||
// interval first; never scan the entire heap in order for a rare term.
|
||||
// If the interval has fewer than size+1 matches, discard it and use
|
||||
// the complete indexed search below. No matches are silently omitted.
|
||||
anchor := cursor.ID
|
||||
if anchor == 0 {
|
||||
if err = tx.QueryRowContext(ctx, r.query(`SELECT coalesce(max(id),0) FROM {schema}papers`)).Scan(&anchor); err != nil {
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
for _, window := range []int64{512, 4096} {
|
||||
bound, op := int64(0), ">="
|
||||
if anchor > window {
|
||||
bound = anchor - window
|
||||
}
|
||||
if cursor.Before {
|
||||
bound, op = int64(9223372036854775807), "<="
|
||||
if anchor <= 9223372036854775807-window {
|
||||
bound = anchor + window
|
||||
}
|
||||
}
|
||||
boundedArgs := append(append([]any{}, args...), bound)
|
||||
bounded := fmt.Sprintf("SELECT %s FROM {schema}papers p%s AND p.id %s $%d ORDER BY %s LIMIT $%d", columns, where, op, len(boundedArgs), order, len(args))
|
||||
positions, err = r.positions(ctx, tx, bounded, boundedArgs)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
// Expand once only when there is evidence of nearby matches.
|
||||
// Zero matches goes straight to the sparse-term index fallback.
|
||||
if len(positions) == 0 || len(positions) >= filters.PageSize+1 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if !topicOnly && len(positions) < filters.PageSize+1 {
|
||||
positions, err = r.positions(ctx, tx, query, args)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
more := len(positions) > filters.PageSize
|
||||
if more {
|
||||
positions = positions[:filters.PageSize]
|
||||
}
|
||||
if cursor.Before {
|
||||
for i, j := 0, len(positions)-1; i < j; i, j = i+1, j-1 {
|
||||
positions[i], positions[j] = positions[j], positions[i]
|
||||
}
|
||||
}
|
||||
if len(positions) > 0 {
|
||||
if (!cursor.Before && cursor.ID != 0) || (cursor.Before && more) {
|
||||
result.PreviousCursor = encodeCursor(original, positions[0], true)
|
||||
}
|
||||
if (!cursor.Before && more) || (cursor.Before && cursor.ID != 0) {
|
||||
result.NextCursor = encodeCursor(original, positions[len(positions)-1], false)
|
||||
}
|
||||
ids := make([]int64, len(positions))
|
||||
byID := map[string]Paper{}
|
||||
for i, p := range positions {
|
||||
ids[i] = p.ID
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, r.query(`SELECT id,title,abstract,authors_json,first_published FROM {schema}papers WHERE id=ANY($1)`), ids)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
for rows.Next() {
|
||||
p, scanErr := r.scanPaper(rows)
|
||||
if scanErr != nil {
|
||||
_ = rows.Close()
|
||||
return result, scanErr
|
||||
}
|
||||
byID[p.ID] = p
|
||||
}
|
||||
err = rows.Err()
|
||||
_ = rows.Close()
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
for _, p := range positions {
|
||||
paper, exists := byID[strconv.FormatInt(p.ID, 10)]
|
||||
if !exists {
|
||||
return result, errors.New("inconsistent paper index")
|
||||
}
|
||||
result.Items = append(result.Items, paper)
|
||||
}
|
||||
}
|
||||
if err = r.enrich(ctx, tx, result.Items); err != nil {
|
||||
return result, err
|
||||
}
|
||||
rows, err = tx.QueryContext(ctx, r.query(`SELECT value FROM {schema}tags WHERE namespace='arxiv:category' ORDER BY value`))
|
||||
rows, err := tx.QueryContext(ctx, r.query(`SELECT value FROM {schema}tags WHERE namespace='arxiv:category' ORDER BY value`))
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
@@ -271,7 +406,7 @@ func (r *PostgresRepository) Get(ctx context.Context, id int64) (Paper, error) {
|
||||
return papers[0], tx.Commit()
|
||||
}
|
||||
|
||||
// Batch metadata queries avoid per-paper round trips and share the count/page
|
||||
// Batch metadata queries avoid per-paper round trips and share the page
|
||||
// snapshot. User values are bound; only the validated schema is interpolated.
|
||||
func (r *PostgresRepository) enrich(ctx context.Context, tx *sql.Tx, papers []Paper) error {
|
||||
if len(papers) == 0 {
|
||||
|
||||
Reference in New Issue
Block a user