This is an automated email from the ASF dual-hosted git repository.
palashc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/phoenix-adapters.git
The following commit(s) were added to refs/heads/main by this push:
new ef44876 PHOENIX-7963 : Support overriding secondary index consistency
via CreateTable Tags (#13)
ef44876 is described below
commit ef4487674641a5c0499caf4f98646a35308416dc
Author: Palash Chauhan <[email protected]>
AuthorDate: Fri Jul 17 22:29:40 2026 -0700
PHOENIX-7963 : Support overriding secondary index consistency via
CreateTable Tags (#13)
---
DDB_API_REFERENCE.md | 45 ++++++++++++++++-
.../phoenix/ddb/service/CreateTableService.java | 34 +++++++++++--
.../phoenix/ddb/service/UpdateTableService.java | 2 +-
.../java/org/apache/phoenix/ddb/CreateTableIT.java | 56 ++++++++++++++++++++++
.../org/apache/phoenix/ddb/TableOptionsConfig.java | 27 ++++++++++-
.../org/apache/phoenix/ddb/utils/ApiMetadata.java | 9 ++++
6 files changed, 166 insertions(+), 7 deletions(-)
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
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 class CreateTableService {
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 class CreateTableService {
+ ") 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 class CreateTableService {
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 class CreateTableService {
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
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 @@ public class UpdateTableService {
= (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 {
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..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
@@ -60,15 +60,20 @@ import
software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException;
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 class CreateTableIT {
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);
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 class TableOptionsConfig {
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 @@ public class TableOptionsConfig {
*/
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";
