OpenSearch

The OpenSearch REST API JSON specification is a collection of JSON files that formalize and standardize the OpenSearch HTTP API. This specification is intended to facilitate the development of client libraries and integrations by providing a machine-readable definition of the API surface.

Each API definition in the specification typically includes:

• API Name: The identifier (e.g., indices.create) which usually corresponds to client library calls.
• Documentation Link: A live URL to the official documentation.
• Stability Level: Indicates the API's lifecycle state.
• Request URL: Defines the HTTP method, path, and path parameters (parts).
• Request Parameters: Defines query or URL parameters.
• Request Body: Defines the schema for the JSON body.

If an API is stable but returns an arbitrary map of key-value pairs, it is noted using the treat_json_as_key_value flag in the response object.

{
  "indices.create": {
    "documentation":{
      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-create-index.html"
    },
    "stability": "stable",
    "url":{
      "paths":[
        {
          "path":"/{index}",
          "method":"PUT",
          "parts":{
            "index":{
              "type":"string",
              "description":"The name of the index"
            }
          }
        }
      ]
    },
    "params": {
      "timeout": {
        "type" : "time",
        "description" : "Explicit operation timeout"
      }
    },
    "body": {
      "description" : "The configuration for the index (`settings` and `mappings`)"
    }
  }
}

The ingestion-hive plugin allows OpenSearch to pull data directly from Apache Hive tables into an index. It connects to a Hive Metastore via Thrift, discovers partitions, and reads Parquet data files from Hadoop-compatible filesystems. This eliminates the need for an intermediate streaming layer like Kafka.

Key capabilities:

• Connects to Hive Metastore using framed or unframed Thrift transport.
• Distributes partitions across shards using consistent hashing.
• Supports incremental partition discovery by monitoring for new partitions.
• Supports Kerberos (SASL/GSSAPI) authentication for the Metastore connection.
• Provides effectively exactly-once delivery guarantees by deriving document _ids from the partition, file, and row index, ensuring idempotent overwrites during recovery.

PUT /my-hive-index
{
  "settings": {
    "ingestion_source": {
      "type": "HIVE",
      "pointer": {
        "init": {
          "reset": "earliest"
        }
      },
      "param": {
        "metastore_uri": "thrift://hive-metastore:9083",
        "database": "my_database",
        "table": "my_table"
      }
    },
    "index.number_of_shards": 3,
    "index.number_of_replicas": 1,
    "index.replication.type": "SEGMENT"
  }
}

When a client receives a response from a server:

• Netty4 Flow: The Netty4TcpChannel receives a BytesReference. The ClientInboundPipeline deserializes it to a TransportResponse (using StreamInput), and the ClientInboundHandler delivers the response to the ResponseHandler.
• Flight Flow (Async Response Handling): The FlightClient uses handleInboundStream(Ticket, Listener) on a FlightClientChannel. This creates a FlightTransportResponse, retrieves the header and reqID, and then uses the reqID to find the correct TransportResponseHandler. The response is then processed asynchronously via handler.handleStreamResponse(streamResponse).

sequenceDiagram
    participant CTC as Client TcpChannel<br/>(Netty4TcpChannel)
    participant CIP as Client InboundPipeline
    participant CIH as Client InboundHandler
    participant RH as ResponseHandler

    Note over CTC,RH: Netty4 Flow
    CTC->>CIP: Receive BytesReference
    CIP->>CIH: Deserialize to TransportResponse<br/>(StreamInput)
    CIH->>RH: Deliver Response

    participant FC as FlightClient
    participant FCC as FlightClientChannel
    participant FTR as FlightTransportResponse
    participant RH2 as ResponseHandler

    Note over FC,RH2: Flight Flow (Async Response Handling)
    FC->>FCC: handleInboundStream(Ticket, Listener)
    FCC->>FTR: Create FlightTransportResponse
    FCC->>FCC: Retrieve Header and reqID
    FCC->>RH2: Get TransportResponseHandler<br/>using reqID
    FCC->>RH2: handler.handleStreamResponse(streamResponse)<br/>(Async Processing)