Build an intelligent metadata management system

Build a multi-dimensional metadata retrieval system with the search index and nested index capabilities of Tablestore.

Solution overview

An intelligent media management platform must efficiently manage metadata for large volumes of files such as images and videos. File metadata includes basic information (file size, creation time, and user) and AI-extracted tag information (tag name and confidence score). The key challenge is that tags are multi-value nested structures. Each file has multiple tags, and each tag has a name and a score. The system must support exact tag-name queries, score-based sorting, and multi-condition retrieval that combines time ranges and file types.

Tablestore search indexes support the nested index type, which creates independent indexes for subfields in JSON arrays. Combine NestedQuery, BoolQuery, and GroupByField to query nested fields, filter by multiple conditions, and aggregate statistics.

Solution design

This solution uses a single data table to store file metadata and a search index for nested tag queries and multi-dimensional retrieval.

Field type	Field	Data type	Index field type	Description
Primary key column	_id	String	—	The MD5 hash of the file ID. This value distributes data evenly and prevents write hotspots.
Attribute column	fId	String	KEYWORD	The file ID, used for exact match queries.
	userId	String	KEYWORD	The user ID, used to filter files by user.
	tags	String (JSON array)	NESTED	An array of AI-recognized tags. Each element contains the subfields `tag` (KEYWORD, the tag name) and `score` (LONG, the confidence score). The nested index type supports NestedQuery on this field.
	type	String	KEYWORD	The file type: image, video, document, or audio.
	size	Integer	LONG	The file size in bytes.
	createdAt	Integer	LONG	The file creation timestamp in milliseconds.
	url	String	KEYWORD	The file storage URL, such as an OSS endpoint.

The tags field stores data as a JSON array, for example [{"tag":"landscape","score":95},{"tag":"architecture","score":80}]. The NESTED type in search indexes automatically parses the JSON array and creates independent indexes for each subfield.

Implementation

Before you begin, activate Tablestore and create an instance. If you already have an available instance, skip this step.

Note

Install and configure Tablestore SDK for Java or Tablestore SDK for Python before you begin. The examples below cover both languages.

Step 1: Create a data table and a search index

When you create the data table, specify only the primary key column. Attribute columns do not need to be predefined. After you create the data table, create a search index that includes a NESTED index with two subfields: tag and score.

Java

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

    TableOptions tableOptions = new TableOptions();
    tableOptions.setMaxVersions(1);
    tableOptions.setTimeToLive(-1);

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

// Create a search index (with nested fields)
private static void createSearchIndex(SyncClient client) {
    CreateSearchIndexRequest request = new CreateSearchIndexRequest();
    request.setTableName("meta_file");
    request.setIndexName("meta_file_index");

    // Sub-field definitions for nested field "tags"
    List<FieldSchema> tagSubFields = Arrays.asList(
        new FieldSchema("tag", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("score", FieldType.LONG).setIndex(true).setEnableSortAndAgg(true)
    );

    IndexSchema indexSchema = new IndexSchema();
    indexSchema.setFieldSchemas(Arrays.asList(
        new FieldSchema("fId", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("userId", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("tags", FieldType.NESTED).setIndex(true).setSubFieldSchemas(tagSubFields),
        new FieldSchema("type", FieldType.KEYWORD).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("size", FieldType.LONG).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("createdAt", FieldType.LONG).setIndex(true).setEnableSortAndAgg(true),
        new FieldSchema("url", FieldType.KEYWORD).setIndex(true)
    ));
    request.setIndexSchema(indexSchema);
    client.createSearchIndex(request);
}

Python

def create_table_and_index(client):
    # Create a data table
    table_meta = tablestore.TableMeta("meta_file", [("_id", "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)

    # Sub-field definitions for nested field "tags"
    tag_sub_fields = [
        tablestore.FieldSchema("tag", tablestore.FieldType.KEYWORD,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("score", tablestore.FieldType.LONG,
                               index=True, enable_sort_and_agg=True),
    ]

    # Create a search index (with nested fields)
    fields = [
        tablestore.FieldSchema("fId", tablestore.FieldType.KEYWORD,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("userId", tablestore.FieldType.KEYWORD,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("tags", tablestore.FieldType.NESTED,
                               index=True, sub_field_schemas=tag_sub_fields),
        tablestore.FieldSchema("type", tablestore.FieldType.KEYWORD,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("size", tablestore.FieldType.LONG,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("createdAt", tablestore.FieldType.LONG,
                               index=True, enable_sort_and_agg=True),
        tablestore.FieldSchema("url", tablestore.FieldType.KEYWORD, index=True),
    ]
    client.create_search_index("meta_file", "meta_file_index",
                               tablestore.SearchIndexMeta(fields))

Step 2: Write test data

Use the BatchWriteRow operation to write 10,000 file metadata records in batches. Each record contains a randomly generated file ID, user ID, file type, AI tags (2 to 4 tags, each with a tag name and confidence score), file size, and creation time. The primary key _id uses the MD5 hash of the file ID to distribute data evenly and prevent write hotspots. BatchWriteRow writes a maximum of 200 rows per request.

Java

private static void writeTestData(SyncClient client) throws Exception {
    Random rand = new Random(42);
    long start = System.currentTimeMillis();
    int written = 0;
    long baseTime = 1704067200000L; // 2024-01-01 UTC
    long timeSpan = 2L * 365 * 24 * 3600 * 1000; // 2 years

    String[] TAGS = {"landscape", "portrait", "architecture", "food", "animal", "plant", "transportation", "sports", "Tablestore", "technology"};
    String[] TYPES = {"image", "video", "document", "audio"};
    String[] USERS = {"u00001", "u00002", "u00003", "u00004", "u00005"};

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

        for (int i = 0; i < batchSize; i++) {
            int idx = written + i;
            String fId = String.format("f%08d", idx);
            String id = md5(fId);
            String userId = USERS[rand.nextInt(USERS.length)];
            String type = TYPES[rand.nextInt(TYPES.length)];
            long size = 1024 + rand.nextInt(10 * 1024 * 1024);
            long createdAt = baseTime + (long)(rand.nextDouble() * timeSpan);
            String url = "oss://bucket/files/" + fId + "." + type;

            // Generate 2-4 random tags with scores
            int tagCount = 2 + rand.nextInt(3);
            Set<Integer> usedIndices = new HashSet<>();
            StringBuilder tagsJson = new StringBuilder("[");
            for (int t = 0; t < tagCount; t++) {
                int tagIdx;
                do { tagIdx = rand.nextInt(TAGS.length); } while (usedIndices.contains(tagIdx));
                usedIndices.add(tagIdx);
                long score = 50 + rand.nextInt(50);
                if (t > 0) tagsJson.append(",");
                tagsJson.append("{\"tag\":\"").append(TAGS[tagIdx])
                        .append("\",\"score\":").append(score).append("}");
            }
            tagsJson.append("]");

            PrimaryKeyBuilder pkb = PrimaryKeyBuilder.createPrimaryKeyBuilder();
            pkb.addPrimaryKeyColumn("_id", PrimaryKeyValue.fromString(id));
            RowPutChange row = new RowPutChange("meta_file", pkb.build());
            row.addColumn("fId", ColumnValue.fromString(fId));
            row.addColumn("userId", ColumnValue.fromString(userId));
            row.addColumn("tags", ColumnValue.fromString(tagsJson.toString()));
            row.addColumn("type", ColumnValue.fromString(type));
            row.addColumn("size", ColumnValue.fromLong(size));
            row.addColumn("createdAt", ColumnValue.fromLong(createdAt));
            row.addColumn("url", ColumnValue.fromString(url));
            batch.addRowChange(row);
        }
        client.batchWriteRow(batch);
        written += batchSize;
    }
    System.out.println("Written 10000 rows in " + (System.currentTimeMillis() - start) + "ms");
}

// MD5 utility method
private static String md5(String input) throws Exception {
    MessageDigest md = MessageDigest.getInstance("MD5");
    byte[] digest = md.digest(input.getBytes("UTF-8"));
    StringBuilder sb = new StringBuilder();
    for (byte b : digest) sb.append(String.format("%02x", b));
    return sb.toString();
}

Python

import hashlib, json, random

def write_test_data(client):
    rand = random.Random(42)
    base_time = 1704067200000  # 2024-01-01 UTC
    time_span = 2 * 365 * 24 * 3600 * 1000  # 2 years

    TAGS = ["landscape", "portrait", "architecture", "food", "animal", "plant", "transportation", "sports", "Tablestore", "technology"]
    TYPES = ["image", "video", "document", "audio"]
    USERS = ["u00001", "u00002", "u00003", "u00004", "u00005"]
    written = 0

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

        for i in range(batch_size):
            idx = written + i
            f_id = f"f{idx:08d}"
            _id = hashlib.md5(f_id.encode()).hexdigest()
            user_id = rand.choice(USERS)
            file_type = rand.choice(TYPES)
            size = 1024 + rand.randint(0, 10 * 1024 * 1024)
            created_at = base_time + int(rand.random() * time_span)
            url = f"oss://bucket/files/{f_id}.{file_type}"

            # Generate 2-4 random tags with scores
            tag_count = 2 + rand.randint(0, 2)
            tag_indices = rand.sample(range(len(TAGS)), tag_count)
            tags_list = [{"tag": TAGS[ti], "score": 50 + rand.randint(0, 49)}
                         for ti in tag_indices]
            tags_json = json.dumps(tags_list, ensure_ascii=False)

            pk = [("_id", _id)]
            cols = [("fId", f_id), ("userId", user_id), ("tags", tags_json),
                    ("type", file_type), ("size", size),
                    ("createdAt", created_at), ("url", url)]
            row = tablestore.Row(pk, cols)
            put_rows.append(tablestore.PutRowItem(
                row, tablestore.Condition(tablestore.RowExistenceExpectation.IGNORE)))

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

Step 3: Query data

After you write data, wait for the search index to synchronize. The examples below cover three typical query scenarios.

Scenario 1: Query files by tag

Use NestedQuery to retrieve all files that contain a specified tag, sorted by creation time in descending order. NestedQuery runs a subquery on nested fields and requires the path to the nested field.

Java

// Query all files with tag "Tablestore"
TermQuery tagQuery = new TermQuery();
tagQuery.setFieldName("tags.tag");
tagQuery.setTerm(ColumnValue.fromString("Tablestore"));

NestedQuery nestedQuery = new NestedQuery();
nestedQuery.setPath("tags");
nestedQuery.setScoreMode(ScoreMode.Avg);
nestedQuery.setQuery(tagQuery);

SearchQuery searchQuery = new SearchQuery();
searchQuery.setQuery(nestedQuery);
searchQuery.setGetTotalCount(true);
searchQuery.setLimit(5);
searchQuery.setSort(new Sort(Arrays.asList(
    new FieldSort("createdAt", SortOrder.DESC))));

SearchRequest request = new SearchRequest("meta_file", "meta_file_index", searchQuery);
SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
columnsToGet.setReturnAll(true);
request.setColumnsToGet(columnsToGet);

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

Python

# Query all files that contain the tag "Tablestore"
nested_query = tablestore.NestedQuery(
    "tags",
    tablestore.TermQuery("tags.tag", "Tablestore"),
    score_mode=tablestore.ScoreMode.AVG,
)

result = client.search("meta_file", "meta_file_index",
    tablestore.SearchQuery(nested_query, get_total_count=True, limit=5,
        sort=tablestore.Sort(sorters=[
            tablestore.FieldSort("createdAt", tablestore.SortOrder.DESC)])),
    tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.ALL))

print(f"Total hits: {result.total_count}")
for row in result.rows:
    cols = {c[0]: c[1] for c in row[1]}
    print(f"fId={cols['fId']} | type={cols['type']} | tags={cols['tags']}")

Scenario 2: Multi-condition query

Use BoolQuery to combine NestedQuery (exact tag match), RangeQuery (time range), and TermQuery (file type filter) for multi-dimensional retrieval. For example, query all image files uploaded in 2024 that contain the tag "Tablestore".

Java

// Nested tag query: tags.tag = "Tablestore"
TermQuery tagQuery = new TermQuery();
tagQuery.setFieldName("tags.tag");
tagQuery.setTerm(ColumnValue.fromString("Tablestore"));

NestedQuery nestedQuery = new NestedQuery();
nestedQuery.setPath("tags");
nestedQuery.setScoreMode(ScoreMode.Avg);
nestedQuery.setQuery(tagQuery);

// Time range query: full year 2024
RangeQuery timeQuery = new RangeQuery();
timeQuery.setFieldName("createdAt");
timeQuery.greaterThanOrEqual(ColumnValue.fromLong(1704067200000L)); // 2024-01-01
timeQuery.lessThan(ColumnValue.fromLong(1735689600000L));           // 2025-01-01

// File type query: type = "image"
TermQuery typeQuery = new TermQuery();
typeQuery.setFieldName("type");
typeQuery.setTerm(ColumnValue.fromString("image"));

// Combine three conditions using BoolQuery
BoolQuery boolQuery = new BoolQuery();
boolQuery.setMustQueries(Arrays.asList(nestedQuery, timeQuery, typeQuery));

SearchQuery searchQuery = new SearchQuery();
searchQuery.setQuery(boolQuery);
searchQuery.setGetTotalCount(true);
searchQuery.setLimit(5);
searchQuery.setSort(new Sort(Arrays.asList(
    new FieldSort("createdAt", SortOrder.DESC))));

SearchRequest request = new SearchRequest("meta_file", "meta_file_index", searchQuery);
SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
columnsToGet.setReturnAll(true);
request.setColumnsToGet(columnsToGet);

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

Python

# Nested tag query
nested_query = tablestore.NestedQuery(
    "tags",
    tablestore.TermQuery("tags.tag", "Tablestore"),
    score_mode=tablestore.ScoreMode.AVG,
)
# Time range query: full year 2024
range_query = tablestore.RangeQuery(
    "createdAt",
    range_from=1704067200000,   # 2024-01-01
    range_to=1735689600000,     # 2025-01-01
    include_lower=True,
    include_upper=False,
)
# File type query
type_query = tablestore.TermQuery("type", "image")

# Combine three conditions using BoolQuery
bool_query = tablestore.BoolQuery(
    must_queries=[nested_query, range_query, type_query])

result = client.search("meta_file", "meta_file_index",
    tablestore.SearchQuery(bool_query, get_total_count=True, limit=5,
        sort=tablestore.Sort(sorters=[
            tablestore.FieldSort("createdAt", tablestore.SortOrder.DESC)])),
    tablestore.ColumnsToGet(return_type=tablestore.ColumnReturnType.ALL))

print(f"Total hits: {result.total_count}")
for row in result.rows:
    cols = {c[0]: c[1] for c in row[1]}
    print(f"fId={cols['fId']} | type={cols['type']} | createdAt={cols['createdAt']}")

Scenario 3: Aggregate by file type

Use GroupByField to group files by type and calculate the count and average file size for each type.

Java

SearchRequest request = SearchRequest.newBuilder()
    .tableName("meta_file").indexName("meta_file_index")
    .searchQuery(SearchQuery.newBuilder()
        .query(QueryBuilders.matchAll())
        .limit(0)
        .addGroupBy(GroupByBuilders.groupByField("fileType", "type")
            .size(20)
            .addSubAggregation(AggregationBuilders.count("fileCount", "fId"))
            .addSubAggregation(AggregationBuilders.avg("avgSize", "size"))
            .addGroupBySorter(GroupBySorter.subAggSortInDesc("fileCount"))
            .build())
        .build())
    .build();

SearchResponse response = client.search(request);
GroupByFieldResult results = response.getGroupByResults()
    .getAsGroupByFieldResult("fileType");
for (GroupByFieldResultItem item : results.getGroupByFieldResultItems()) {
    long count = item.getSubAggregationResults()
        .getAsCountAggregationResult("fileCount").getValue();
    double avgSize = item.getSubAggregationResults()
        .getAsAvgAggregationResult("avgSize").getValue();
    System.out.println("Type: " + item.getKey()
        + " | File count: " + count
        + " | Avg size: " + String.format("%.0f", avgSize) + " bytes");
}

Python

group_by = tablestore.GroupByField("type", size=20,
    sub_aggs=[
        tablestore.Count("fId", name="fileCount"),
        tablestore.Avg("size", name="avgSize"),
    ],
    group_by_sort=[tablestore.SubAggSort(tablestore.SortOrder.DESC, "fileCount")],
    name="fileType")

result = client.search("meta_file", "meta_file_index",
    tablestore.SearchQuery(tablestore.MatchAllQuery(), 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"Type: {item.key} | File count: {int(aggs['fileCount'])}"
          f" | Average size: {aggs['avgSize']:.0f} bytes")

Clean up resources

Note

If you no longer need the resources created in this solution, follow these steps to clean them up and avoid unnecessary costs.

Delete the search index first, and then delete the data table. If you created the instance specifically for this solution, release the instance last.

Java

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

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

Python

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