PHOENIX-7963 : Support overriding secondary index consistency via CreateTable Tags

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
 

phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/CreateTableService.java

@@ -62,7 +62,7 @@ public static Map<String, Object> getCreateTableResponse(final String tableName,
 
     public static void addIndexDDL(String tableName, List<Map<String, Object>> keySchemaElements,
             List<Map<String, Object>> attributeDefinitions, List<String> 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<Map<String, Object>> keySc
                 + ") WHERE " + indexHashKey + " IS NOT " + "NULL " + ((indexSortKey
                 != null) ? " AND " + indexSortKey + " IS NOT " + "NULL " : "") + (isAsync ?
                 " ASYNC " :
-                "") + TableOptionsConfig.getIndexOptions());
+                "") + TableOptionsConfig.getIndexOptions(indexConsistency));
     }
 
     public static List<String> getIndexDDLs(Map<String, Object> request) {
         final List<String> indexDDLs = new ArrayList<>();
         List<Map<String, Object>> attributeDefinitions =
                 (List<Map<String, Object>>) request.get(ApiMetadata.ATTRIBUTE_DEFINITIONS);
+        final String indexConsistency = resolveIndexConsistency(request);
 
         if (request.get(ApiMetadata.GLOBAL_SECONDARY_INDEXES) != null) {
             for (Map<String, Object> globalSecondaryIndex : (List<Map<String, Object>>) request.get(
@@ -178,7 +179,7 @@ public static List<String> getIndexDDLs(Map<String, Object> request) {
                 final List<Map<String, Object>> keySchemaElements =
                         (List<Map<String, Object>>) 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<String> getIndexDDLs(Map<String, Object> request) {
                 final List<Map<String, Object>> keySchemaElements =
                         (List<Map<String, Object>>) 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<String, Object> request) {
+        List<Map<String, Object>> tags =
+                (List<Map<String, Object>>) request.get(ApiMetadata.TAGS);
+        if (tags == null) {
+            return null;
+        }
+        for (Map<String, Object> 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

phoenix-ddb-rest/src/main/java/org/apache/phoenix/ddb/service/UpdateTableService.java

@@ -85,7 +85,7 @@ private static void handleIndexUpdates(Map<String, Object> request, String table
                             = (List<Map<String, Object>>) request.get(ApiMetadata.ATTRIBUTE_DEFINITIONS);
                     List<Map<String, Object>> keySchema
                             = (List<Map<String, Object>>) 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 {

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,57 @@ 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;
+        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,
+            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);

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(".")) {

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";