From 66635a341085891c087e6dda36fe9b4e784568d6 Mon Sep 17 00:00:00 2001 From: Palash Chauhan Date: Fri, 17 Jul 2026 16:05:36 -0700 Subject: [PATCH 1/3] PHOENIX-7963 : Support overriding secondary index consistency via CreateTable Tags Co-authored-by: Cursor --- .../ddb/service/CreateTableService.java | 34 ++++++++++-- .../ddb/service/UpdateTableService.java | 2 +- .../org/apache/phoenix/ddb/CreateTableIT.java | 52 +++++++++++++++++++ .../phoenix/ddb/TableOptionsConfig.java | 27 +++++++++- .../apache/phoenix/ddb/utils/ApiMetadata.java | 9 ++++ 5 files changed, 118 insertions(+), 6 deletions(-) diff --git a/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/CreateTableService.java b/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/CreateTableService.java index b68712b..8d452c6 100644 --- a/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/CreateTableService.java +++ b/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/CreateTableService.java @@ -62,7 +62,7 @@ public static Map getCreateTableResponse(final String tableName, public static void addIndexDDL(String tableName, List> keySchemaElements, List> attributeDefinitions, List indexDDLs, - String indexName, boolean isAsync) { + String indexName, boolean isAsync, String indexConsistency) { final StringBuilder indexOn = new StringBuilder(); String indexHashKey = null; @@ -163,13 +163,14 @@ public static void addIndexDDL(String tableName, List> keySc + ") WHERE " + indexHashKey + " IS NOT " + "NULL " + ((indexSortKey != null) ? " AND " + indexSortKey + " IS NOT " + "NULL " : "") + (isAsync ? " ASYNC " : - "") + TableOptionsConfig.getIndexOptions()); + "") + TableOptionsConfig.getIndexOptions(indexConsistency)); } public static List getIndexDDLs(Map request) { final List indexDDLs = new ArrayList<>(); List> attributeDefinitions = (List>) request.get(ApiMetadata.ATTRIBUTE_DEFINITIONS); + final String indexConsistency = resolveIndexConsistency(request); if (request.get(ApiMetadata.GLOBAL_SECONDARY_INDEXES) != null) { for (Map globalSecondaryIndex : (List>) request.get( @@ -178,7 +179,7 @@ public static List getIndexDDLs(Map request) { final List> keySchemaElements = (List>) globalSecondaryIndex.get(ApiMetadata.KEY_SCHEMA); addIndexDDL((String)request.get(ApiMetadata.TABLE_NAME), keySchemaElements, - attributeDefinitions, indexDDLs, indexName, false); + attributeDefinitions, indexDDLs, indexName, false, indexConsistency); } } @@ -189,12 +190,37 @@ public static List getIndexDDLs(Map request) { final List> keySchemaElements = (List>) localSecondaryIndex.get(ApiMetadata.KEY_SCHEMA); addIndexDDL((String)request.get(ApiMetadata.TABLE_NAME), keySchemaElements, - attributeDefinitions, indexDDLs, indexName, false); + attributeDefinitions, indexDDLs, indexName, false, indexConsistency); } } return indexDDLs; } + /** + * Resolves the table-wide secondary index consistency from request Tags. Returns + * {@code STRONG} when the {@code phoenix.index.consistency} tag requests it, or null to fall + * back to the configured default. Only STRONG is accepted as an override value. + */ + static String resolveIndexConsistency(Map request) { + List> tags = + (List>) request.get(ApiMetadata.TAGS); + if (tags == null) { + return null; + } + for (Map tag : tags) { + if (ApiMetadata.TAG_INDEX_CONSISTENCY.equals(tag.get(ApiMetadata.TAG_KEY))) { + String value = (String) tag.get(ApiMetadata.TAG_VALUE); + if (ApiMetadata.INDEX_CONSISTENCY_STRONG.equalsIgnoreCase(value)) { + return ApiMetadata.INDEX_CONSISTENCY_STRONG; + } + throw new ValidationException("Unsupported value '" + value + "' for tag " + + ApiMetadata.TAG_INDEX_CONSISTENCY + "; only " + + ApiMetadata.INDEX_CONSISTENCY_STRONG + " is supported"); + } + } + return null; + } + /** * If StreamEnabled is set to true, return 2 DDLs for CDC. * 1. CREATE CDC ddl to create the virtual cdc table and index diff --git a/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/UpdateTableService.java b/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/UpdateTableService.java index 4cda95a..de6761e 100644 --- a/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/UpdateTableService.java +++ b/phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/UpdateTableService.java @@ -85,7 +85,7 @@ private static void handleIndexUpdates(Map request, String table = (List>) request.get(ApiMetadata.ATTRIBUTE_DEFINITIONS); List> keySchema = (List>) createIndexUpdate.get(ApiMetadata.KEY_SCHEMA); - CreateTableService.addIndexDDL(tableName, keySchema, attrDefs, indexDDLs, indexName, true); + CreateTableService.addIndexDDL(tableName, keySchema, attrDefs, indexDDLs, indexName, true, null); LOGGER.info("DDL for Create Index: {}", indexDDLs); indexDDLs.addAll(ddl); } else { diff --git a/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java b/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java index fa165ed..6b95145 100644 --- a/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java +++ b/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java @@ -60,15 +60,20 @@ import software.amazon.awssdk.services.dynamodb.model.ResourceInUseException; import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType; import software.amazon.awssdk.services.dynamodb.model.TableDescription; +import software.amazon.awssdk.services.dynamodb.model.Tag; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.phoenix.ddb.rest.RESTServer; +import org.apache.phoenix.ddb.utils.ApiMetadata; import org.apache.phoenix.end2end.ServerMetadataCacheTestImpl; import org.apache.phoenix.exception.PhoenixIOException; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixDriver; import org.apache.phoenix.schema.PColumn; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.PTableKey; +import org.apache.phoenix.schema.types.IndexConsistency; import org.apache.phoenix.util.PhoenixRuntime; import org.apache.phoenix.util.ServerUtil; @@ -318,6 +323,53 @@ public void createTableWithStreamTest() throws Exception { TestUtils.validateCdcProps(url, tableName); } + @Test(timeout = 120000) + public void createTableWithStrongConsistencyTag() throws Exception { + final String tableName = testName.getMethodName().toUpperCase(); + CreateTableRequest createTableRequest = + DDLTestUtils.getCreateTableRequest(tableName, "PK1", ScalarAttributeType.B, "PK2", + ScalarAttributeType.S); + final String gsi = "IDX1_" + tableName; + final String lsi = "IDX2_" + tableName; + createTableRequest = DDLTestUtils.addIndexToRequest(true, createTableRequest, gsi, "COL1", + ScalarAttributeType.N, "COL2", ScalarAttributeType.B); + createTableRequest = DDLTestUtils.addIndexToRequest(false, createTableRequest, lsi, "PK1", + ScalarAttributeType.B, "LCOL2", ScalarAttributeType.S); + createTableRequest = createTableRequest.toBuilder().tags(Tag.builder() + .key(ApiMetadata.TAG_INDEX_CONSISTENCY).value(ApiMetadata.INDEX_CONSISTENCY_STRONG) + .build()).build(); + + phoenixDBClientV2.createTable(createTableRequest); + + assertIndexConsistency(tableName, gsi, IndexConsistency.STRONG); + assertIndexConsistency(tableName, lsi, IndexConsistency.STRONG); + } + + @Test(timeout = 120000) + public void createTableDefaultsToEventualConsistency() throws Exception { + final String tableName = testName.getMethodName().toUpperCase(); + CreateTableRequest createTableRequest = + DDLTestUtils.getCreateTableRequest(tableName, "PK1", ScalarAttributeType.B, "PK2", + ScalarAttributeType.S); + final String gsi = "IDX1_" + tableName; + createTableRequest = DDLTestUtils.addIndexToRequest(true, createTableRequest, gsi, "COL1", + ScalarAttributeType.N, "COL2", ScalarAttributeType.B); + + phoenixDBClientV2.createTable(createTableRequest); + + assertIndexConsistency(tableName, gsi, IndexConsistency.EVENTUAL); + } + + private static void assertIndexConsistency(String tableName, String indexName, + IndexConsistency expected) throws Exception { + String fullIndexName = PhoenixUtils.getFullTableName(tableName + "_" + indexName, false); + try (Connection connection = DriverManager.getConnection(url)) { + PhoenixConnection pconn = connection.unwrap(PhoenixConnection.class); + PTable index = pconn.getTable(new PTableKey(pconn.getTenantId(), fullIndexName)); + Assert.assertEquals(expected, index.getIndexConsistency()); + } + } + @Test(timeout = 120000) public void createTableTest() throws Exception { createTable(dynamoDbClient); diff --git a/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/TableOptionsConfig.java b/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/TableOptionsConfig.java index af16df0..28b5be0 100644 --- a/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/TableOptionsConfig.java +++ b/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/TableOptionsConfig.java @@ -13,9 +13,12 @@ public class TableOptionsConfig { private static final String TABLE_CONFIG_FILE = "phoenix-table-options.properties"; private static final String INDEX_CONFIG_FILE = "phoenix-index-options.properties"; private static final String UPDATE_CACHE_FREQUENCY_KEY = "UPDATE_CACHE_FREQUENCY"; + private static final String CONSISTENCY_KEY = "CONSISTENCY"; private static String tableOptionsString; private static String indexOptionsString; + private static String indexBaseOptionsString; + private static String defaultIndexConsistency; private static String cdcOptionsString; /** @@ -24,6 +27,10 @@ public class TableOptionsConfig { public static void initialize() throws IOException { tableOptionsString = buildOptionsString(TABLE_CONFIG_FILE, "table"); indexOptionsString = buildOptionsString(INDEX_CONFIG_FILE, "index"); + Properties indexProps = loadConfiguration(INDEX_CONFIG_FILE, "index"); + defaultIndexConsistency = indexProps.getProperty(CONSISTENCY_KEY); + indexProps.remove(CONSISTENCY_KEY); + indexBaseOptionsString = joinOptions(indexProps); cdcOptionsString = buildCdcOptionsString(TABLE_CONFIG_FILE); LOGGER.info("Initialized table, index, and CDC configurations"); } @@ -48,6 +55,22 @@ public static String getIndexOptions() { return indexOptionsString; } + /** + * Get index options with an optional CONSISTENCY override. When {@code consistencyOverride} + * is null the configured default consistency is used. + */ + public static String getIndexOptions(String consistencyOverride) { + if (indexBaseOptionsString == null) { + throw new IllegalStateException("Index Options Config not initialized."); + } + String consistency = + consistencyOverride != null ? consistencyOverride : defaultIndexConsistency; + if (consistency == null) { + return indexBaseOptionsString; + } + return indexBaseOptionsString + "," + CONSISTENCY_KEY + "=" + consistency; + } + /** * Get CDC options as formatted string for CREATE CDC statements. */ @@ -75,8 +98,10 @@ private static String buildCdcOptionsString(String tableConfigFile) throws IOExc */ private static String buildOptionsString(String configFile, String configType) throws IOException { - Properties props = loadConfiguration(configFile, configType); + return joinOptions(loadConfiguration(configFile, configType)); + } + private static String joinOptions(Properties props) { return String.join(",", props.stringPropertyNames().stream().map(key -> { // Handle quoted properties for HBase/Phoenix specific options if (key.contains(".")) { diff --git a/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/utils/ApiMetadata.java b/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/utils/ApiMetadata.java index de67a34..e8551f9 100644 --- a/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/utils/ApiMetadata.java +++ b/phoenix-ddb-utils/src/main/java/org/apache/phoenix/ddb/utils/ApiMetadata.java @@ -53,6 +53,15 @@ public class ApiMetadata { public static final String GLOBAL_SECONDARY_INDEXES = "GlobalSecondaryIndexes"; public static final String GLOBAL_SECONDARY_INDEX_UPDATES = "GlobalSecondaryIndexUpdates"; + // ---------- Tags ---------- + public static final String TAGS = "Tags"; + public static final String TAG_KEY = "Key"; + public static final String TAG_VALUE = "Value"; + // Table-wide override of secondary index consistency. Only STRONG is accepted; when absent + // the configured default (EVENTUAL) applies. + public static final String TAG_INDEX_CONSISTENCY = "phoenix.index.consistency"; + public static final String INDEX_CONSISTENCY_STRONG = "STRONG"; + // ---------- UpdateTable ---------- public static final String CREATE = "Create"; public static final String DELETE = "Delete"; From 6937030e495480c704c27643335e7c8721a977d2 Mon Sep 17 00:00:00 2001 From: Palash Chauhan Date: Fri, 17 Jul 2026 16:16:13 -0700 Subject: [PATCH 2/3] PHOENIX-7963 : Document CreateTable Tags for secondary index consistency override Co-authored-by: Cursor --- DDB_API_REFERENCE.md | 45 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/DDB_API_REFERENCE.md b/DDB_API_REFERENCE.md index 9495e8b..e6742ed 100644 --- a/DDB_API_REFERENCE.md +++ b/DDB_API_REFERENCE.md @@ -439,6 +439,7 @@ Creates a new table in Phoenix with the specified key schema, attributes, option | `GlobalSecondaryIndexes` | List | No | Global secondary indexes to create | | `LocalSecondaryIndexes` | List | No | Local secondary indexes to create | | `StreamSpecification` | Map | No | Enable change data capture stream | +| `Tags` | List | No | Key/value tags. Recognized tags can override table-level behavior (see [Tags](#tags)) | **KeySchema element structure:** ```json @@ -484,6 +485,38 @@ Creates a new table in Phoenix with the specified key schema, attributes, option - `StreamViewType` values: `NEW_IMAGE`, `OLD_IMAGE`, `NEW_AND_OLD_IMAGES` - Required when `StreamEnabled` is `true` +#### Tags + +`Tags` is an optional list of key/value pairs. Specific recognized tag keys let you override +table-level behavior at creation time. + +**Tags element structure:** +```json +{ + "Tags": [ + {"Key": "phoenix.index.consistency", "Value": "STRONG"} + ] +} +``` + +**Recognized tags:** + +| Tag Key | Allowed Values | Default | Effect | +|---|---|---|---| +| `phoenix.index.consistency` | `STRONG` | `EVENTUAL` | Overrides the consistency of **all** secondary indexes (GSIs and LSIs) created in this `CreateTable` request | + +**Secondary index consistency (`phoenix.index.consistency`):** + +- By default, secondary indexes are created with **eventual** consistency. +- Setting the `phoenix.index.consistency` tag to `STRONG` makes every GSI and LSI defined in + the same `CreateTable` request **strongly** consistent. The override applies table-wide to all + indexes in that request; per-index granularity is not supported. +- The value is case-insensitive. Only `STRONG` is accepted as an override — any other value + (e.g. `EVENTUAL`, `weak`) throws `400` with a `ValidationException`. To use eventual + consistency, simply omit the tag. +- The tag only affects indexes created by this request. Indexes added later via `UpdateTable` + are created with the configured default consistency. + #### Response ```json @@ -510,6 +543,7 @@ Creates a new table in Phoenix with the specified key schema, attributes, option - All key attributes in `KeySchema` must have a matching `AttributeDefinitions` entry - Attribute types must be `S`, `N`, or `B` - If `StreamEnabled` is `true`, `StreamViewType` must be non-empty +- If the `phoenix.index.consistency` tag is present, its value must be `STRONG` (case-insensitive); any other value throws 400 with `ValidationException` - If the table already exists (created more than 5 seconds ago), throws 400 with `ResourceInUseException` #### Phoenix SQL Generated @@ -539,6 +573,15 @@ CREATE UNCOVERED INDEX IF NOT EXISTS "status-index" WHERE BSON_VALUE("COL", 'status', 'VARCHAR') IS NOT NULL ``` +When the `phoenix.index.consistency` tag is set to `STRONG`, a `CONSISTENCY=STRONG` option is +appended to the index DDL for every index in the request: +```sql +CREATE UNCOVERED INDEX IF NOT EXISTS "status-index" + ON "SCHEMA"."MyTable" (BSON_VALUE("COL", 'status', 'VARCHAR'), BSON_VALUE("COL", 'created_at', 'DOUBLE')) + WHERE BSON_VALUE("COL", 'status', 'VARCHAR') IS NOT NULL + CONSISTENCY=STRONG +``` + --- ### 6.2 DeleteTable @@ -1867,7 +1910,7 @@ Each API operation tracks: | **Stream shard iterators** | Expire after 15 minutes | No automatic expiry | | **KCL consumer compatibility** | KCL | KCL compatible (via `dynamodb-streams-kinesis-adapter`) | | **Item storage** | Native DynamoDB format | BSON document in a single Phoenix column | -| **Consistency** | Eventual + (Strong for local indexes) | Depends on Phoenix/HBase configuration | +| **Consistency** | Eventual + (Strong for local indexes) | Secondary indexes default to eventual; can be made strong table-wide at CreateTable via the `phoenix.index.consistency=STRONG` tag | ### Key Schema Constraints From 795befe2aead41e4684d735b413561fc71cdbe79 Mon Sep 17 00:00:00 2001 From: Palash Chauhan Date: Fri, 17 Jul 2026 16:34:42 -0700 Subject: [PATCH 3/3] PHOENIX-7963 : Assert default eventual consistency for both GSI and LSI Co-authored-by: Cursor --- .../src/test/java/org/apache/phoenix/ddb/CreateTableIT.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java b/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java index 6b95145..bd5ad34 100644 --- a/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java +++ b/phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/CreateTableIT.java @@ -352,12 +352,16 @@ public void createTableDefaultsToEventualConsistency() throws Exception { DDLTestUtils.getCreateTableRequest(tableName, "PK1", ScalarAttributeType.B, "PK2", ScalarAttributeType.S); final String gsi = "IDX1_" + tableName; + final String lsi = "IDX2_" + tableName; createTableRequest = DDLTestUtils.addIndexToRequest(true, createTableRequest, gsi, "COL1", ScalarAttributeType.N, "COL2", ScalarAttributeType.B); + createTableRequest = DDLTestUtils.addIndexToRequest(false, createTableRequest, lsi, "PK1", + ScalarAttributeType.B, "LCOL2", ScalarAttributeType.S); phoenixDBClientV2.createTable(createTableRequest); assertIndexConsistency(tableName, gsi, IndexConsistency.EVENTUAL); + assertIndexConsistency(tableName, lsi, IndexConsistency.EVENTUAL); } private static void assertIndexConsistency(String tableName, String indexName,