Build a large-scale insurance policy query system

Tablestore search indexes let you build a large-scale insurance policy query system that supports multi-condition queries, nested queries, and aggregations.

Overview

Insurance businesses generate large volumes of policy data. Policyholders, brokers, and insurance companies need to search and analyze policies across multiple dimensions — for example, querying active policies for a specific policyholder, aggregating premiums by product or broker, and running nested queries on beneficiary information.

A single Tablestore table supports PB-scale storage and tens of millions of TPS, making it well-suited for large-scale policy data. Search indexes, built on inverted indexes and columnar storage, support multi-condition queries, range queries, nested queries, and aggregations. In production, policy transaction data is typically written to a relational database such as MySQL and synchronized to Tablestore in real time through Data Transmission Service (DTS) to separate reads from writes.

Design

A single data table stores all policy data, and a search index supports various query types. The following design uses life insurance policies as an example.

Field type	Field	Data type	Index field type	Description
Primary key	policy_id_md5	STRING	KEYWORD	MD5 hash of the policy ID. Used as the partition key to evenly distribute data.
Attribute columns	policy_id	STRING	KEYWORD	Globally unique policy ID.
	product_name	STRING	KEYWORD	Insurance product name.
	operate_time	STRING	DATE	Application date.
	effective_time	STRING	DATE	Policy effective date.
	expiration_time	STRING	DATE	Policy expiration date.
	applier_user_id	STRING	KEYWORD	Policyholder user ID.
	broker_user_id	STRING	KEYWORD	Broker user ID.
	applier_name	STRING	KEYWORD	Policyholder name.
	applier_id	STRING	KEYWORD	Policyholder ID number.
	applier_gender	STRING	KEYWORD	Policyholder gender.
	insured_name	STRING	KEYWORD	Insured person's name.
	insured_id	STRING	KEYWORD	Insured person's ID number.
	insured_gender	STRING	KEYWORD	Insured person's gender.
	beneficiary_info	STRING	NESTED	Beneficiary information (JSON array).
	premium	DOUBLE	DOUBLE	Premium amount.
	profit	DOUBLE	DOUBLE	Sum insured.

beneficiary_info is stored as a JSON array string in the data table and mapped to the NESTED type in the search index. The nested fields have the following structure:

Field	Type	Description
name	KEYWORD	Beneficiary name
id	KEYWORD	Beneficiary ID number
gender	KEYWORD	Beneficiary gender
benefit_percentage	LONG	Benefit percentage

Sample beneficiary_info data:

[
  {"name": "Emily", "id": "198905050697770441", "gender": "female", "benefit_percentage": 4},
  {"name": "Sarah", "id": "198905050697770441", "gender": "female", "benefit_percentage": 96}
]

Implementation

Before you begin, activate Tablestore and create an instance. Skip this if you already have one. For more information, see Activate Tablestore and create an instance.

Note

The following examples use Java and Python. Install and configure Java SDK or Python SDK before you proceed.

Step 1: Create a data table and a search index

When you create a data table, specify only the primary key columns — attribute columns do not need to be predefined. After the data table is created, create a search index for it.

Java

// Create a data table
private static void createTable(SyncClient client) {
    TableMeta tableMeta = new TableMeta("policy");
    tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema("policy_id_md5", PrimaryKeyType.STRING));

    TableOptions tableOptions = new TableOptions();
    tableOptions.setMaxVersions(1);
    tableOptions.setTimeToLive(-1); // Data never expires

    CreateTableRequest request = new CreateTableRequest(tableMeta, tableOptions);
    client.createTable(request);
}

// Create a search index
private static void createSearchIndex(SyncClient client) {
    CreateSearchIndexRequest request = new CreateSearchIndexRequest();
    request.setTableName("policy");
    request.setIndexName("policy_index");

    // Beneficiary nested fields
    List<FieldSchema> subFields = new ArrayList<>();
    subFields.add(new FieldSchema("name", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true));
    subFields.add(new FieldSchema("id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true));
    subFields.add(new FieldSchema("gender", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true));
    subFields.add(new FieldSchema("benefit_percentage", FieldType.LONG).setIndex(true).setEnableSortAndAgg(true));

    IndexSchema indexSchema = new IndexSchema();
    indexSchema.setFieldSchemas(Arrays.asList(
        new FieldSchema("policy_id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("product_name", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("operate_time", FieldType.DATE).setIndex(true).setEnableSortAndAgg(true)
            .setDateFormats(Arrays.asList("yyyy-MM-dd HH:mm:ss")),
        new FieldSchema("effective_time", FieldType.DATE).setIndex(true).setEnableSortAndAgg(true)
            .setDateFormats(Arrays.asList("yyyy-MM-dd HH:mm:ss")),
        new FieldSchema("expiration_time", FieldType.DATE).setIndex(true).setEnableSortAndAgg(true)
            .setDateFormats(Arrays.asList("yyyy-MM-dd HH:mm:ss")),
        new FieldSchema("applier_user_id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("broker_user_id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("applier_name", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("applier_id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("applier_gender", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("insured_name", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("insured_id", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("insured_gender", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("premium", FieldType.DOUBLE).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("profit", FieldType.DOUBLE).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("beneficiary_info", FieldType.NESTED).setSubFieldSchemas(subFields)
    ));

    request.setIndexSchema(indexSchema);
    client.createSearchIndex(request);
}

Python

# Create a data table
def create_table(client):
    table_meta = tablestore.TableMeta("policy", [("policy_id_md5", "STRING")])
    table_options = tablestore.TableOptions(time_to_live=-1, max_version=1)
    reserved = tablestore.ReservedThroughput(tablestore.CapacityUnit(0, 0))
    client.create_table(table_meta, table_options, reserved)

# Create a search index
def create_search_index(client):
    nested_sub_fields = [
        tablestore.FieldSchema("name", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("gender", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("benefit_percentage", tablestore.FieldType.LONG, index=True, enable_sort_and_agg=True),
    ]
    fields = [
        tablestore.FieldSchema("policy_id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("product_name", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("operate_time", tablestore.FieldType.DATE, index=True, enable_sort_and_agg=True,
            date_formats=["yyyy-MM-dd HH:mm:ss"]),
        tablestore.FieldSchema("effective_time", tablestore.FieldType.DATE, index=True, enable_sort_and_agg=True,
            date_formats=["yyyy-MM-dd HH:mm:ss"]),
        tablestore.FieldSchema("expiration_time", tablestore.FieldType.DATE, index=True, enable_sort_and_agg=True,
            date_formats=["yyyy-MM-dd HH:mm:ss"]),
        tablestore.FieldSchema("applier_user_id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("broker_user_id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("applier_name", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("applier_id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("applier_gender", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("insured_name", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("insured_id", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("insured_gender", tablestore.FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("premium", tablestore.FieldType.DOUBLE, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("profit", tablestore.FieldType.DOUBLE, index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("beneficiary_info", tablestore.FieldType.NESTED, sub_field_schemas=nested_sub_fields),
    ]
    client.create_search_index("policy", "policy_index", tablestore.SearchIndexMeta(fields))

Step 2: Write test data

In production, tools such as DTS, DataX, and Canal synchronize policy data from MySQL to Tablestore in real time. This example uses the BatchWriteRow API to write randomly generated test data.

The following code generates 10,000 policy records covering 3 insurance products and 3 brokers. Policy expiration dates are split between 2025 and 2027 to support different query types. Adjust TOTAL_ROWS to change the data volume.

Java

private static final int TOTAL_ROWS = 10000;
private static final String[] PRODUCTS = {"Accident Insurance", "Annuity Insurance", "Life Insurance"};
private static final String[] BROKERS = {"broker_001", "broker_002", "broker_003"};
private static final String[] NAMES = {"John", "Lisa", "David", "Sarah", "Michael", "Emily", "James", "Anna"};

private static void writeTestData(SyncClient client) throws Exception {
    Random rand = new Random(42);
    DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    int written = 0;

    while (written < TOTAL_ROWS) {
        BatchWriteRowRequest batch = new BatchWriteRowRequest();
        int batchSize = Math.min(200, TOTAL_ROWS - written);

        for (int i = 0; i < batchSize; i++) {
            int idx = written + i;
            String policyId = String.format("policy-%06d", idx);
            // Use MD5 hash as the partition key for even data distribution
            String policyIdMd5 = md5(policyId);
            String product = PRODUCTS[rand.nextInt(PRODUCTS.length)];
            String broker = BROKERS[rand.nextInt(BROKERS.length)];

            // Application date: random within 2024
            int dayOffset = rand.nextInt(365);
            LocalDateTime operateTime = LocalDateTime.of(2024, 1, 1, 0, 0)
                .plusDays(dayOffset).plusHours(rand.nextInt(24)).plusMinutes(rand.nextInt(60));
            LocalDateTime effectiveTime = operateTime.plusDays(5 + rand.nextInt(10));

            // Expiration: even-indexed rows expire in August 2025 (for expiration queries),
            // odd-indexed rows expire in 2027+ (for active policy queries)
            LocalDateTime expirationTime;
            if (idx % 2 == 0) {
                expirationTime = LocalDateTime.of(2025, 8, 1 + rand.nextInt(28),
                    rand.nextInt(24), rand.nextInt(60));
            } else {
                expirationTime = effectiveTime.plusYears(3 + rand.nextInt(2));
            }

            String applierId = String.format("1%017d", 100000000000000L + idx);
            double premium = (500 + rand.nextInt(10) * 500);
            double profit = premium * (5 + rand.nextInt(16));

            // Beneficiary information: 1–2 beneficiaries with percentages that sum to 100
            int pct1 = 10 + rand.nextInt(81);
            StringBuilder benJson = new StringBuilder("[");
            benJson.append(String.format(
                "{\"id\": \"%s\", \"name\": \"%s\", \"gender\": \"%s\", \"benefit_percentage\": %d}",
                applierId, NAMES[rand.nextInt(NAMES.length)],
                rand.nextBoolean() ? "male" : "female", pct1));
            if (rand.nextBoolean()) {
                benJson.append(String.format(
                    ", {\"id\": \"%s\", \"name\": \"%s\", \"gender\": \"%s\", \"benefit_percentage\": %d}",
                    applierId, NAMES[rand.nextInt(NAMES.length)],
                    rand.nextBoolean() ? "male" : "female", 100 - pct1));
            }
            benJson.append("]");

            PrimaryKeyBuilder pkBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
            pkBuilder.addPrimaryKeyColumn("policy_id_md5", PrimaryKeyValue.fromString(policyIdMd5));
            RowPutChange row = new RowPutChange("policy", pkBuilder.build());
            row.addColumn("policy_id", ColumnValue.fromString(policyId));
            row.addColumn("product_name", ColumnValue.fromString(product));
            row.addColumn("operate_time", ColumnValue.fromString(operateTime.format(fmt)));
            row.addColumn("effective_time", ColumnValue.fromString(effectiveTime.format(fmt)));
            row.addColumn("expiration_time", ColumnValue.fromString(expirationTime.format(fmt)));
            row.addColumn("applier_user_id", ColumnValue.fromString("user-" + idx));
            row.addColumn("broker_user_id", ColumnValue.fromString(broker));
            row.addColumn("applier_name", ColumnValue.fromString(NAMES[rand.nextInt(NAMES.length)]));
            row.addColumn("applier_id", ColumnValue.fromString(applierId));
            row.addColumn("applier_gender", ColumnValue.fromString(rand.nextBoolean() ? "male" : "female"));
            row.addColumn("insured_name", ColumnValue.fromString(NAMES[rand.nextInt(NAMES.length)]));
            row.addColumn("insured_id", ColumnValue.fromString(applierId));
            row.addColumn("insured_gender", ColumnValue.fromString(rand.nextBoolean() ? "male" : "female"));
            row.addColumn("beneficiary_info", ColumnValue.fromString(benJson.toString()));
            row.addColumn("premium", ColumnValue.fromDouble(premium));
            row.addColumn("profit", ColumnValue.fromDouble(profit));
            batch.addRowChange(row);
        }
        client.batchWriteRow(batch);
        written += batchSize;
    }
    System.out.println("Wrote " + TOTAL_ROWS + " rows.");
}

Python

import hashlib, json, random
from datetime import datetime, timedelta

TOTAL_ROWS = 10000
PRODUCTS = ["Accident Insurance", "Annuity Insurance", "Life Insurance"]
BROKERS = ["broker_001", "broker_002", "broker_003"]
NAMES = ["John", "Lisa", "David", "Sarah", "Michael", "Emily", "James", "Anna"]

def write_test_data(client):
    rand = random.Random(42)
    base_date = datetime(2024, 1, 1)
    written = 0
    fmt = "%Y-%m-%d %H:%M:%S"

    while written < TOTAL_ROWS:
        batch_size = min(200, TOTAL_ROWS - written)
        put_rows = []

        for i in range(batch_size):
            idx = written + i
            policy_id = f"policy-{idx:06d}"
            policy_id_md5 = hashlib.md5(policy_id.encode()).hexdigest()

            operate_time = base_date + timedelta(days=rand.randint(0, 364),
                hours=rand.randint(0, 23), minutes=rand.randint(0, 59))
            effective_time = operate_time + timedelta(days=rand.randint(5, 14))
            if idx % 2 == 0:
                expiration_time = datetime(2025, 8, rand.randint(1, 28),
                    rand.randint(0, 23), rand.randint(0, 59))
            else:
                expiration_time = effective_time + timedelta(days=365 * rand.randint(3, 4))

            applier_id = f"1{100000000000000 + idx:017d}"
            premium = rand.randint(1, 10) * 500.0
            pct1 = rand.randint(10, 90)
            bens = [{"id": applier_id, "name": rand.choice(NAMES),
                     "gender": rand.choice(["male", "female"]), "benefit_percentage": pct1}]
            if rand.choice([True, False]):
                bens.append({"id": applier_id, "name": rand.choice(NAMES),
                    "gender": rand.choice(["male", "female"]), "benefit_percentage": 100 - pct1})

            pk = [("policy_id_md5", policy_id_md5)]
            cols = [
                ("policy_id", policy_id), ("product_name", rand.choice(PRODUCTS)),
                ("operate_time", operate_time.strftime(fmt)),
                ("effective_time", effective_time.strftime(fmt)),
                ("expiration_time", expiration_time.strftime(fmt)),
                ("applier_user_id", f"user-{idx}"), ("broker_user_id", rand.choice(BROKERS)),
                ("applier_name", rand.choice(NAMES)), ("applier_id", applier_id),
                ("applier_gender", rand.choice(["male", "female"])),
                ("insured_name", rand.choice(NAMES)), ("insured_id", applier_id),
                ("insured_gender", rand.choice(["male", "female"])),
                ("beneficiary_info", json.dumps(bens, ensure_ascii=False)),
                ("premium", premium), ("profit", premium * rand.randint(5, 20)),
            ]
            row = tablestore.Row(pk, cols)
            condition = tablestore.Condition(tablestore.RowExistenceExpectation.IGNORE)
            put_rows.append(tablestore.PutRowItem(row, condition))

        request = tablestore.BatchWriteRowRequest()
        request.add(tablestore.TableInBatchWriteRowItem("policy", put_rows))
        client.batch_write_row(request)
        written += batch_size

    print(f"Wrote {TOTAL_ROWS} rows.")

Step 3: Query data

The search index synchronizes data automatically after writes, typically within seconds. The following examples cover five common policy query scenarios.

Scenario 1: Query active policies of a specific policyholder

Combine TermQuery (exact match on the policyholder ID number) with RangeQuery (effective date ≤ current time < expiration date) to find active policies.

Java

public static void queryEffectivePolicy(SyncClient client) {
    String nowStr = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

    TermQuery termQuery = new TermQuery();
    termQuery.setFieldName("applier_id");
    termQuery.setTerm(ColumnValue.fromString("100100000000000001"));

    RangeQuery rangeQuery1 = new RangeQuery();
    rangeQuery1.setFieldName("effective_time");
    rangeQuery1.lessThanOrEqual(ColumnValue.fromString(nowStr));

    RangeQuery rangeQuery2 = new RangeQuery();
    rangeQuery2.setFieldName("expiration_time");
    rangeQuery2.greaterThan(ColumnValue.fromString(nowStr));

    BoolQuery boolQuery = new BoolQuery();
    boolQuery.setMustQueries(Arrays.asList(termQuery, rangeQuery1, rangeQuery2));

    SearchQuery searchQuery = new SearchQuery();
    searchQuery.setQuery(boolQuery);
    searchQuery.setGetTotalCount(true);

    SearchRequest searchRequest = new SearchRequest("policy", "policy_index", searchQuery);
    SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
    columnsToGet.setReturnAll(true);
    searchRequest.setColumnsToGet(columnsToGet);

    SearchResponse response = client.search(searchRequest);
    System.out.println("Total matches: " + response.getTotalCount());
    for (Row row : response.getRows()) {
        System.out.println(row);
    }
}

Python

from datetime import datetime

def query_effective_policy(client):
    now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    bool_query = tablestore.BoolQuery(must_queries=[
        tablestore.TermQuery("applier_id", "100100000000000001"),
        tablestore.RangeQuery("effective_time", range_to=now_str, include_upper=True),
        tablestore.RangeQuery("expiration_time", range_from=now_str, include_lower=False),
    ])

    result = client.search("policy", "policy_index",
        tablestore.SearchQuery(bool_query, get_total_count=True),
        tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.ALL))

    print(f"Total matches: {result.total_count}")
    for row in result.rows:
        print(row)

Scenario 2: Query active policies for a specific beneficiary with a benefit percentage above 10%

Extend Scenario 1 by adding a NestedQuery on the beneficiary_info field. The nested query requires the beneficiary ID number and the percentage threshold to be met within the same sub-row.

Java

public static void queryBeneficiaryInfo(SyncClient client) {
    String nowStr = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

    // Nested query: match both the beneficiary ID and benefit_percentage >= 10 within the same sub-row
    TermQuery nestedTermQuery = new TermQuery();
    nestedTermQuery.setFieldName("beneficiary_info.id");
    nestedTermQuery.setTerm(ColumnValue.fromString("100100000000000001"));

    RangeQuery nestedRangeQuery = new RangeQuery();
    nestedRangeQuery.setFieldName("beneficiary_info.benefit_percentage");
    nestedRangeQuery.greaterThanOrEqual(ColumnValue.fromLong(10));

    BoolQuery nestedBoolQuery = new BoolQuery();
    nestedBoolQuery.setMustQueries(Arrays.asList(nestedTermQuery, nestedRangeQuery));

    NestedQuery nestedQuery = new NestedQuery();
    nestedQuery.setPath("beneficiary_info");
    nestedQuery.setQuery(nestedBoolQuery);
    nestedQuery.setScoreMode(ScoreMode.None);

    // Outer conditions: currently active
    RangeQuery rangeQuery1 = new RangeQuery();
    rangeQuery1.setFieldName("effective_time");
    rangeQuery1.lessThanOrEqual(ColumnValue.fromString(nowStr));

    RangeQuery rangeQuery2 = new RangeQuery();
    rangeQuery2.setFieldName("expiration_time");
    rangeQuery2.greaterThan(ColumnValue.fromString(nowStr));

    BoolQuery boolQuery = new BoolQuery();
    boolQuery.setMustQueries(Arrays.asList(rangeQuery1, rangeQuery2, nestedQuery));

    SearchQuery searchQuery = new SearchQuery();
    searchQuery.setQuery(boolQuery);
    searchQuery.setGetTotalCount(true);

    SearchRequest searchRequest = new SearchRequest("policy", "policy_index", searchQuery);
    SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
    columnsToGet.setReturnAll(true);
    searchRequest.setColumnsToGet(columnsToGet);

    SearchResponse response = client.search(searchRequest);
    System.out.println("Total matches: " + response.getTotalCount());
    for (Row row : response.getRows()) {
        System.out.println(row);
    }
}

Python

def query_beneficiary_info(client):
    now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    nested_query = tablestore.NestedQuery("beneficiary_info",
        tablestore.BoolQuery(must_queries=[
            tablestore.TermQuery("beneficiary_info.id", "100100000000000001"),
            tablestore.RangeQuery("beneficiary_info.benefit_percentage",
                range_from=10, include_lower=True),
        ]))

    bool_query = tablestore.BoolQuery(must_queries=[
        tablestore.RangeQuery("effective_time", range_to=now_str, include_upper=True),
        tablestore.RangeQuery("expiration_time", range_from=now_str, include_lower=False),
        nested_query,
    ])

    result = client.search("policy", "policy_index",
        tablestore.SearchQuery(bool_query, get_total_count=True),
        tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.ALL))

    print(f"Total matches: {result.total_count}")
    for row in result.rows:
        print(row)

Scenario 3: Query policies expiring within a specific time range

Use RangeQuery to find policies expiring in August 2025. This is useful for renewal reminders.

Java

public static void queryExpirationPolicy(SyncClient client) {
    RangeQuery rangeQuery = new RangeQuery();
    rangeQuery.setFieldName("expiration_time");
    rangeQuery.greaterThan(ColumnValue.fromString("2025-08-01 00:00:00"));
    rangeQuery.lessThan(ColumnValue.fromString("2025-09-01 00:00:00"));

    SearchQuery searchQuery = new SearchQuery();
    searchQuery.setQuery(rangeQuery);
    searchQuery.setGetTotalCount(true);
    searchQuery.setLimit(5);

    SearchRequest searchRequest = new SearchRequest("policy", "policy_index", searchQuery);
    SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
    columnsToGet.setColumns(Arrays.asList("policy_id", "product_name", "expiration_time", "premium"));
    searchRequest.setColumnsToGet(columnsToGet);

    SearchResponse response = client.search(searchRequest);
    System.out.println("Total matches: " + response.getTotalCount());
    for (Row row : response.getRows()) {
        System.out.println(row);
    }
}

Python

def query_expiration_policy(client):
    result = client.search("policy", "policy_index",
        tablestore.SearchQuery(
            tablestore.RangeQuery("expiration_time",
                range_from="2025-08-01 00:00:00", include_lower=False,
                range_to="2025-09-01 00:00:00", include_upper=False),
            get_total_count=True, limit=5),
        tablestore.ColumnsToGet(["policy_id", "product_name", "expiration_time", "premium"],
            return_type=tablestore.ColumnReturnType.SPECIFIED))

    print(f"Total matches: {result.total_count}")
    for row in result.rows:
        print(row)

Scenario 4: Aggregate policy count and total premiums by product

Use GroupByField to group by product name, with Count and Sum sub-aggregations to calculate the policy count and total premium for each product. Results are sorted by total premium in descending order.

Java

public static void groupByProduct(SyncClient client) {
    SearchRequest searchRequest = SearchRequest.newBuilder()
        .tableName("policy").indexName("policy_index")
        .searchQuery(SearchQuery.newBuilder()
            .query(QueryBuilders.range("operate_time")
                .greaterThanOrEqual("2024-01-01 00:00:00")
                .lessThan("2025-01-01 00:00:00"))
            .limit(0)
            .addGroupBy(GroupByBuilders.groupByField("productName", "product_name")
                .size(20)
                .addSubAggregation(AggregationBuilders.count("policyCount", "policy_id"))
                .addSubAggregation(AggregationBuilders.sum("policyPremium", "premium"))
                .addGroupBySorter(GroupBySorter.subAggSortInDesc("policyPremium"))
                .build())
            .build())
        .build();

    SearchResponse resp = client.search(searchRequest);
    GroupByFieldResult results = resp.getGroupByResults().getAsGroupByFieldResult("productName");
    for (GroupByFieldResultItem item : results.getGroupByFieldResultItems()) {
        System.out.println("Product: " + item.getKey()
            + "\tPolicy count: " + item.getSubAggregationResults().getAsCountAggregationResult("policyCount").getValue()
            + "\tTotal premium: " + item.getSubAggregationResults().getAsSumAggregationResult("policyPremium").getValue());
    }
}

Python

def group_by_product(client):
    query = tablestore.RangeQuery("operate_time",
        range_from="2024-01-01 00:00:00", include_lower=True,
        range_to="2025-01-01 00:00:00", include_upper=False)

    group_by = tablestore.GroupByField("product_name", size=20,
        sub_aggs=[
            tablestore.Count("policy_id", name="policyCount"),
            tablestore.Sum("premium", name="policyPremium"),
        ],
        group_by_sort=[tablestore.SubAggSort(tablestore.SortOrder.DESC, "policyPremium")],
        name="productName")

    result = client.search("policy", "policy_index",
        tablestore.SearchQuery(query, limit=0, group_bys=[group_by]),
        tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.NONE))

    for item in result.group_by_results[0].items:
        aggs = {a.name: a.value for a in item.sub_aggs}
        print(f"Product: {item.key}\tPolicy count: {int(aggs['policyCount'])}\tTotal premium: {aggs['policyPremium']}")

Scenario 5: Aggregate policy count and total premiums by broker

Use GroupByField to group by broker ID and calculate performance metrics per broker.

Java

public static void groupByBroker(SyncClient client) {
    SearchRequest searchRequest = SearchRequest.newBuilder()
        .tableName("policy").indexName("policy_index")
        .searchQuery(SearchQuery.newBuilder()
            .query(QueryBuilders.range("operate_time")
                .greaterThanOrEqual("2024-01-01 00:00:00")
                .lessThan("2025-01-01 00:00:00"))
            .limit(0)
            .addGroupBy(GroupByBuilders.groupByField("broker", "broker_user_id")
                .size(20)
                .addSubAggregation(AggregationBuilders.count("policyCount", "policy_id"))
                .addSubAggregation(AggregationBuilders.sum("policyPremium", "premium"))
                .addGroupBySorter(GroupBySorter.subAggSortInDesc("policyPremium"))
                .build())
            .build())
        .build();

    SearchResponse resp = client.search(searchRequest);
    GroupByFieldResult results = resp.getGroupByResults().getAsGroupByFieldResult("broker");
    for (GroupByFieldResultItem item : results.getGroupByFieldResultItems()) {
        System.out.println("Broker: " + item.getKey()
            + "\tPolicy count: " + item.getSubAggregationResults().getAsCountAggregationResult("policyCount").getValue()
            + "\tTotal premium: " + item.getSubAggregationResults().getAsSumAggregationResult("policyPremium").getValue());
    }
}

Python

def group_by_broker(client):
    query = tablestore.RangeQuery("operate_time",
        range_from="2024-01-01 00:00:00", include_lower=True,
        range_to="2025-01-01 00:00:00", include_upper=False)

    group_by = tablestore.GroupByField("broker_user_id", size=20,
        sub_aggs=[
            tablestore.Count("policy_id", name="policyCount"),
            tablestore.Sum("premium", name="policyPremium"),
        ],
        group_by_sort=[tablestore.SubAggSort(tablestore.SortOrder.DESC, "policyPremium")],
        name="broker")

    result = client.search("policy", "policy_index",
        tablestore.SearchQuery(query, limit=0, group_bys=[group_by]),
        tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.NONE))

    for item in result.group_by_results[0].items:
        aggs = {a.name: a.value for a in item.sub_aggs}
        print(f"Broker: {item.key}\tPolicy count: {int(aggs['policyCount'])}\tTotal premium: {aggs['policyPremium']}")

Clean up resources

Note

To avoid unnecessary costs, delete the resources created in this tutorial if you no longer need them.

Delete resources in this order: the search index, then the data table. If you created the instance for this tutorial, release the instance last.

Java

private static void cleanup(SyncClient client) {
    // 1. Delete the search index
    DeleteSearchIndexRequest deleteIndexReq = new DeleteSearchIndexRequest();
    deleteIndexReq.setTableName("policy");
    deleteIndexReq.setIndexName("policy_index");
    client.deleteSearchIndex(deleteIndexReq);

    // 2. Delete the data table
    DeleteTableRequest deleteTableReq = new DeleteTableRequest("policy");
    client.deleteTable(deleteTableReq);
}

Python

def cleanup(client):
    # 1. Delete the search index
    client.delete_search_index("policy", "policy_index")
    # 2. Delete the data table
    client.delete_table("policy")