Elasticsearch Adapter

[palcarazm]

Short Description of the Feature

Implement an Elasticsearch adapter (ElasticsearchBatchEntityReader and ElasticsearchBatchEntityWriter) for BatchJS-Data. Uses the official Elasticsearch Node.js client with DSL JSON for queries and _bulk API for writes. Supports from/size pagination with optional search_after for deep pagination.

Expected Benefits

• Search/analytics use cases: Elasticsearch is a popular search and analytics engine.
• DSL flexibility: Users can use full Elasticsearch DSL for complex queries.
• Bulk operations: _bulk API is efficient for large writes.

Acceptance Criteria

• src/elasticsearch/classes/ElasticsearchBatchEntityReader.ts implements AbstractNoSQLBatchEntityReaderStream using @elastic/elasticsearch.
• src/elasticsearch/classes/ElasticsearchBatchEntityWriter.ts implements AbstractNoSQLBatchEntityWriterStream using @elastic/elasticsearch.
• Query is DSL JSON object (e.g., { match: { title: 'search' } }).
• Pagination uses from/size (default) with optional search_after for deep pagination.
• Connection is managed via Client instance passed in options.
• Batch writes use _bulk API.
• No transaction support (documented limitation).
• @elastic/elasticsearch is an optional peer dependency in package.json (^8.x).
• ElasticsearchBatchEntityReaderOptions<T> and ElasticsearchBatchEntityWriterOptions<T> interfaces exist.
• The adapter is exported from src/elasticsearch/index.ts.
• JSDoc documentation covers all classes, methods, and options.
• npm run docs:elasticsearch generates docs/elasticsearch-api.md.
• Test coverage >=80% (unit + integration tests using testcontainers for Elasticsearch).
• README.md includes an Elasticsearch usage example.
• TypeScript types are published correctly for batchjs-data/elasticsearch.

Documentation

API Design

ElasticsearchBatchEntityReader

export interface ElasticsearchBatchEntityReaderOptions<T> {
  client: Client;               // Elasticsearch client instance
  index: string;                // Index name
  query: QueryDslQueryContainer; // Elasticsearch DSL JSON query
  paginationMode?: 'from' | 'search_after'; // Pagination mode
  sort?: Record<string, 'asc' | 'desc'>;    // Sort field for search_after
  rowToEntity: (hit: any) => T; // Convert Elasticsearch hit to entity
  batchSize: number;            // Batch size
}

export class ElasticsearchBatchEntityReader<T> extends AbstractNoSQLBatchEntityReaderStream<T> {
  constructor(options: ElasticsearchBatchEntityReaderOptions<T>);
}

ElasticsearchBatchEntityWriter

export interface ElasticsearchBatchEntityWriterOptions<T> {
  client: Client;               // Elasticsearch client instance
  index: string;                // Index name
  entityToDocument: (entity: T) => any; // Convert entity to document
  batchSize: number;            // Batch size
}

export class ElasticsearchBatchEntityWriter<T> extends AbstractNoSQLBatchEntityWriterStream<T> {
  constructor(options: ElasticsearchBatchEntityWriterOptions<T>);
}

Usage Example

import { Client } from "@elastic/elasticsearch";
import { ElasticsearchBatchEntityReader, ElasticsearchBatchEntityWriter } from "batchjs-data/elasticsearch";

const client = new Client({ node: "http://localhost:9200" });

class UserBatchReader extends ElasticsearchBatchEntityReader<UserDTO> {
  constructor(batchSize: number) {
    super({
      batchSize,
      client,
      index: "users",
      query: { match: { active: true } },
      paginationMode: "search_after",
      sort: { id: "asc" },
      rowToEntity: (hit) => ({ id: hit._source.id, username: hit._source.name })
    });
  }
}

class UserBatchWriter extends ElasticsearchBatchEntityWriter<UserDTO> {
  constructor(batchSize: number) {
    super({
      batchSize,
      client,
      index: "users",
      entityToDocument: (entity) => ({ id: entity.id, name: entity.username })
    });
  }
}

Pagination Implementation

from/size:

const response = await this.client.search({
  index: this.index,
  body: {
    query: this.query,
    from: this.from,
    size: size
  }
});
this.from += size;
return response.hits.hits.map(this.rowToEntity);

search_after:

const body: any = {
  query: this.query,
  size: size,
  sort: [this.sort]
};
if (this.searchAfter) {
  body.search_after = this.searchAfter;
}
const response = await this.client.search({ index: this.index, body });
const hits = response.hits.hits;
if (hits.length > 0) {
  this.searchAfter = hits[hits.length - 1].sort;
}
return hits.map(this.rowToEntity);

Batch Write Implementation

const operations: any[] = [];
for (const entity of chunk) {
  operations.push({ index: { _index: this.index } });
  operations.push(this.entityToDocument(entity));
}
await this.client.bulk({ operations });

Driver Compatibility

• @elastic/elasticsearch version: ^8.0.0
• Elasticsearch versions: 7.17, 8.x (as supported by driver)
• Connection handling: Uses client instance directly

Additional Comments

• from/size pagination has a default limit of 10,000 documents (index.max_result_window). For large datasets, users should use search_after.
• Elasticsearch _bulk API accepts operations and documents in alternating array format.
• No transaction support — _bulk is atomic at the index level but not transactional.

Feature Request Checklist

• Confirm that you agree to follow the project's code of conduct.
• Confirm that you have reviewed open and rejected feature requests to ensure novelty.