This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 6b8bbbb658 PHOENIX-7584 Conditional TTL on SYSTEM.CDC_STREAM (#2211)
6b8bbbb658 is described below
commit 6b8bbbb658338d393c001a415d504ad5d20851fc
Author: Palash Chauhan <[email protected]>
AuthorDate: Wed Jul 9 10:25:25 2025 -0700
PHOENIX-7584 Conditional TTL on SYSTEM.CDC_STREAM (#2211)
---
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 ++
.../phoenix/query/ConnectionQueryServicesImpl.java | 8 ++++-
.../org/apache/phoenix/query/QueryServices.java | 2 ++
.../apache/phoenix/query/QueryServicesOptions.java | 2 ++
.../org/apache/phoenix/schema/MetaDataClient.java | 27 ++++++++-------
.../java/org/apache/phoenix/util/MetaDataUtil.java | 21 ++++++++++--
.../java/org/apache/phoenix/util/ScanUtil.java | 7 ++--
.../apache/phoenix/end2end/BasePermissionsIT.java | 6 ++++
.../org/apache/phoenix/end2end/CDCStreamIT.java | 38 +++++++++++++++++++++-
9 files changed, 95 insertions(+), 18 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 3625dd006e..c9cde56ca3 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -468,6 +468,8 @@ public class PhoenixDatabaseMetaData implements
DatabaseMetaData {
public static final String PARTITION_START_KEY = "PARTITION_START_KEY";
public static final String PARTITION_END_KEY = "PARTITION_END_KEY";
public static final String PARENT_PARTITION_START_TIME =
"PARENT_PARTITION_START_TIME";
+ public static final String CDC_STREAM_CONDITIONAL_TTL_EXPRESSION =
"PARTITION_END_TIME IS NOT NULL " +
+ "AND TO_NUMBER(CURRENT_TIME()) -
TO_NUMBER(PHOENIX_ROW_TIMESTAMP()) >= %d";
public static final String QUERY_ID = "QUERY_ID";
public static final String USER = "USER";
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 11d672c3b2..9d2f9c6975 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -32,6 +32,7 @@ import static
org.apache.phoenix.coprocessorclient.MetaDataProtocol.PHOENIX_MINO
import static
org.apache.phoenix.coprocessorclient.MetaDataProtocol.PHOENIX_PATCH_NUMBER;
import static org.apache.phoenix.coprocessorclient.MetaDataProtocol.getVersion;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ARRAY_SIZE;
+import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CDC_STREAM_CONDITIONAL_TTL_EXPRESSION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_DEF;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
@@ -3872,7 +3873,12 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices implement
}
protected String getCDCStreamDDL() {
- return
setSystemDDLProperties(QueryConstants.CREATE_CDC_STREAM_METADATA);
+ long partitionExpiryMinAge = getProps().getLong(
+ QueryServices.PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS,
+
QueryServicesOptions.DEFAULT_PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS);
+ String ttlExpression =
String.format(CDC_STREAM_CONDITIONAL_TTL_EXPRESSION, partitionExpiryMinAge);
+ String ddl =
setSystemDDLProperties(QueryConstants.CREATE_CDC_STREAM_METADATA);
+ return ddl + ",TTL='" + ttlExpression + "'";
}
private String setSystemDDLProperties(String ddl) {
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
index e9d0e8d86e..00480c2dd4 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -560,6 +560,8 @@ public interface QueryServices extends SQLCloseable {
String CQSI_THREAD_POOL_METRICS_ENABLED =
"phoenix.cqsi.thread.pool.metrics.enabled";
+ String PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS =
"phoenix.cdc.stream.partition.expiry.min.age.ms";
+
/**
* Get executor service used for parallel scans
*/
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index b0ab07d901..70f47adfa7 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -471,6 +471,8 @@ public class QueryServicesOptions {
public static final Boolean
DEFAULT_CQSI_THREAD_POOL_ALLOW_CORE_THREAD_TIMEOUT = true;
public static final Boolean DEFAULT_CQSI_THREAD_POOL_METRICS_ENABLED =
false;
public static final int DEFAULT_CDC_TTL_MUTATION_MAX_RETRIES = 5;
+
+ public static final long
DEFAULT_PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS = 30*60*60*1000; // 30
hours
private final Configuration config;
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 5b9498644f..4cc466ad21 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -36,6 +36,7 @@ import static
org.apache.phoenix.query.QueryServices.INDEX_CREATE_DEFAULT_STATE;
import static org.apache.phoenix.schema.PTableType.CDC;
import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_FOREVER;
import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_NOT_DEFINED;
+import static org.apache.phoenix.schema.PTableType.SYSTEM;
import static
org.apache.phoenix.thirdparty.com.google.common.collect.Sets.newLinkedHashSet;
import static
org.apache.phoenix.thirdparty.com.google.common.collect.Sets.newLinkedHashSetWithExpectedSize;
import static
org.apache.phoenix.exception.SQLExceptionCode.INSUFFICIENT_MULTI_TENANT_COLUMNS;
@@ -1080,7 +1081,7 @@ public class MetaDataClient {
Map<String,Object> tableProps =
Maps.newHashMapWithExpectedSize(statement.getProps().size());
Map<String,Object> commonFamilyProps =
Maps.newHashMapWithExpectedSize(statement.getProps().size() + 1);
populatePropertyMaps(statement.getProps(), tableProps,
commonFamilyProps,
- statement.getTableType(), false);
+ statement.getTableType(),
SchemaUtil.getUnEscapedFullName(tableName.toString()), false);
splits = processSplits(tableProps, splits);
boolean isAppendOnlySchema = false;
@@ -1236,7 +1237,7 @@ public class MetaDataClient {
* @throws SQLException
*/
private void populatePropertyMaps(ListMultimap<String,Pair<String,Object>>
statementProps, Map<String, Object> tableProps,
- Map<String, Object> commonFamilyProps, PTableType tableType,
boolean isCDCIndex) throws SQLException {
+ Map<String, Object> commonFamilyProps, PTableType tableType,
String tableName, boolean isCDCIndex) throws SQLException {
// Somewhat hacky way of determining if property is for
HColumnDescriptor or HTableDescriptor
ColumnFamilyDescriptor defaultDescriptor =
ColumnFamilyDescriptorBuilder.of(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES);
if (!statementProps.isEmpty()) {
@@ -1249,7 +1250,9 @@ public class MetaDataClient {
.buildException();
}
// Handle when TTL property is set
- if (prop.getFirst().equalsIgnoreCase(TTL) && tableType !=
PTableType.SYSTEM) {
+ if (prop.getFirst().equalsIgnoreCase(TTL)
+ && (tableType != PTableType.SYSTEM
+ ||
MetaDataUtil.SYSTEM_TABLES_WITH_TTL_SUPPORTED.contains(tableName))) {
tableProps.put(prop.getFirst(), prop.getSecond());
if (prop.getSecond() != null) {
TTLExpression ttlExpr =
MetaDataUtil.convertForeverAndNoneTTLValue(prop.getSecond(), false);
@@ -1637,7 +1640,7 @@ public class MetaDataClient {
Map<String,Object> tableProps =
Maps.newHashMapWithExpectedSize(statement.getProps().size());
Map<String,Object> commonFamilyProps =
Maps.newHashMapWithExpectedSize(statement.getProps().size() + 1);
- populatePropertyMaps(statement.getProps(), tableProps,
commonFamilyProps, PTableType.INDEX,
+ populatePropertyMaps(statement.getProps(), tableProps,
commonFamilyProps, PTableType.INDEX, indexTableName.toString(),
CDCUtil.isCDCIndex(SchemaUtil
.getTableNameFromFullName(statement.getIndexTableName().toString())));
List<Pair<ParseNode, SortOrder>> indexParseNodeAndSortOrderList =
ik.getParseNodeAndSortOrderList();
@@ -1982,7 +1985,7 @@ public class MetaDataClient {
Map<String, Object> commonFamilyProps =
Maps.newHashMapWithExpectedSize(
statement.getProps().size() + 1);
populatePropertyMaps(statement.getProps(), tableProps,
commonFamilyProps, PTableType.CDC,
- false);
+ cdcObjName, false);
Properties props = connection.getClientInfo();
props.put(INDEX_CREATE_DEFAULT_STATE, "ACTIVE");
@@ -2495,6 +2498,7 @@ public class MetaDataClient {
tableNameNode = statement.getTableName();
final String schemaName = connection.getSchema() != null &&
tableNameNode.getSchemaName() == null ? connection.getSchema() :
tableNameNode.getSchemaName();
final String tableName = tableNameNode.getTableName();
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableName);
String parentTableName = null;
PName tenantId = connection.getTenantId();
String tenantIdStr = tenantId == null ? null :
tenantId.getString();
@@ -2536,7 +2540,7 @@ public class MetaDataClient {
.buildException();
}
- if (tableType != TABLE && (tableType != VIEW || viewType !=
UPDATABLE)) {
+ if (!MetaDataUtil.isTTLSupported(tableType, viewType,
fullTableName)) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.
TTL_SUPPORTED_FOR_TABLES_AND_VIEWS_ONLY)
.setSchemaName(schemaName)
@@ -2563,7 +2567,7 @@ public class MetaDataClient {
.build()
.buildException();
}
- ttl = getCompatibleTTLExpression(ttlProp, tableType);
+ ttl = getCompatibleTTLExpression(ttlProp, tableType, viewType,
fullTableName);
} else {
ttlFromHierarchy = checkAndGetTTLFromHierarchy(parent,
tableName);
if (!ttlFromHierarchy.equals(TTL_EXPRESSION_NOT_DEFINED)) {
@@ -6212,8 +6216,8 @@ public class MetaDataClient {
.buildException();
}
- if (table.getType() != PTableType.TABLE && (table.getType() !=
PTableType.VIEW ||
- table.getViewType() != UPDATABLE)) {
+ if (!MetaDataUtil.isTTLSupported(
+ table.getType(), table.getViewType(),
table.getName().toString())) {
throw new SQLExceptionInfo.Builder(
SQLExceptionCode.TTL_SUPPORTED_FOR_TABLES_AND_VIEWS_ONLY)
.build()
@@ -6223,7 +6227,8 @@ public class MetaDataClient {
TTLExpression newTTL = metaProperties.getTTL();
newTTL.validateTTLOnAlter(connection, table);
metaPropertiesEvaluated.setTTL(
- getCompatibleTTLExpression(metaProperties.getTTL(),
table.getType()));
+ getCompatibleTTLExpression(metaProperties.getTTL(),
table.getType(),
+ table.getViewType(),
table.getName().toString()));
changingPhoenixTableProperty = true;
}
//Updating Introducing TTL variable to true so that we will check
if TTL is already
@@ -6613,7 +6618,7 @@ public class MetaDataClient {
} else if (changePermsStatement.getTableName() != null) {
PTable inputTable = connection.getTable(SchemaUtil.
normalizeFullTableName(changePermsStatement.getTableName().toString()));
- if (!(PTableType.TABLE.equals(inputTable.getType()) ||
PTableType.SYSTEM.equals(inputTable.getType()))) {
+ if (!(PTableType.TABLE.equals(inputTable.getType()) ||
SYSTEM.equals(inputTable.getType()))) {
throw new AccessDeniedException("Cannot GRANT or REVOKE
permissions on INDEX TABLES or VIEWS");
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
index e3d5377e3b..1fe8798898 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
@@ -21,6 +21,7 @@ import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.*;
import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_DEFINED_IN_TABLE_DESCRIPTOR;
import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_FOREVER;
import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_NOT_DEFINED;
+import static org.apache.phoenix.schema.PTableType.SYSTEM;
import static org.apache.phoenix.schema.PTableType.TABLE;
import static org.apache.phoenix.schema.PTableType.VIEW;
import static org.apache.phoenix.util.SchemaUtil.getVarChars;
@@ -124,6 +125,10 @@ public class MetaDataUtil {
ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS,
ColumnFamilyDescriptorBuilder.REPLICATION_SCOPE);
+ public static final List<String> SYSTEM_TABLES_WITH_TTL_SUPPORTED
+ = Arrays.asList(SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA,
SYSTEM_CDC_STREAM_TABLE),
+ SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA,
SYSTEM_CDC_STREAM_STATUS_TABLE));
+
public static Put getLastDDLTimestampUpdate(byte[] tableHeaderRowKey,
long clientTimestamp,
long lastDDLTimestamp) {
@@ -1229,9 +1234,12 @@ public class MetaDataUtil {
* if expression == TTL_NOT_DEFINED OR FOREVER then
* TTL_NOT_DEFINED will be stores in SYSTEM.CATALOG
*/
- public static TTLExpression getCompatibleTTLExpression(TTLExpression
expression, PTableType tableType)
+ public static TTLExpression getCompatibleTTLExpression(TTLExpression
expression,
+ PTableType
tableType,
+ PTable.ViewType
viewType,
+ String tableName)
throws SQLException {
- if (tableType != TABLE && tableType != VIEW) {
+ if (!isTTLSupported(tableType, viewType, tableName)) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.
TTL_SUPPORTED_FOR_TABLES_AND_VIEWS_ONLY)
.build()
@@ -1248,4 +1256,13 @@ public class MetaDataUtil {
return ttl;
}
+ /**
+ * TTL is supported only for tables, updatable views and few SYSTEM tables
+ * defined in SYSTEM_TABLES_WITH_TTL_SUPPORTED.
+ */
+ public static boolean isTTLSupported(PTableType tableType, PTable.ViewType
viewType, String tableName) {
+ return tableType == TABLE
+ || (tableType == VIEW && viewType == PTable.ViewType.UPDATABLE)
+ || (tableType == SYSTEM &&
SYSTEM_TABLES_WITH_TTL_SUPPORTED.contains(tableName));
+ }
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java
index 1941692f57..86710eb51b 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java
@@ -1413,12 +1413,13 @@ public class ScanUtil {
return;
}
- // If is a system table or If Phoenix level TTL/compaction is not
enabled
+ // If is a system table with TTL not supported or If Phoenix level
TTL/compaction is not enabled
// then set the TTL scan attribute to TTL_DEFINED_IN_TABLE_DESCRIPTOR.
if
((!isPhoenixCompactionEnabled(phoenixConnection.getQueryServices().getConfiguration()))
||
- SchemaUtil.isSystemTable(
+ (SchemaUtil.isSystemTable(
SchemaUtil.getTableNameAsBytes(table.getSchemaName().getString(),
- table.getTableName().getString()))) {
+ table.getTableName().getString()))
+ &&
!MetaDataUtil.SYSTEM_TABLES_WITH_TTL_SUPPORTED.contains(table.getName().getString())))
{
byte[] ttlForScan =
TTL_EXPRESSION_DEFINED_IN_TABLE_DESCRIPTOR.serialize();
scan.setAttribute(BaseScannerRegionObserverConstants.TTL,
ttlForScan);
return;
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
index 53e8cb6894..9f09c39f2d 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
@@ -117,6 +117,10 @@ public abstract class BasePermissionsIT extends BaseTest {
QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\""
+ PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME + "\"";
+ private static final String SYSTEM_CDC_STREAM_IDENTIFIER =
+ QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\""
+ + PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_TABLE + "\"";
+
static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES = new
HashSet<>(Arrays.asList(
"SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS",
"SYSTEM:FUNCTION", "SYSTEM:MUTEX",
"SYSTEM:CHILD_LINK","SYSTEM:TRANSFORM",
"SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM"));
@@ -1070,6 +1074,7 @@ public abstract class BasePermissionsIT extends BaseTest {
}
verifyAllowed(grantPermissions("RWX", user,
SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
verifyAllowed(grantPermissions("RWX", user,
SYSTEM_MUTEX_IDENTIFIER, false), superUser);
+ verifyAllowed(grantPermissions("RWX", user,
SYSTEM_CDC_STREAM_IDENTIFIER, false), superUser);
}
}
@@ -1079,6 +1084,7 @@ public abstract class BasePermissionsIT extends BaseTest {
verifyAllowed(revokePermissions(user,
QueryConstants.SYSTEM_SCHEMA_NAME, true), superUser);
verifyAllowed(revokePermissions(user,
SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
verifyAllowed(revokePermissions(user, SYSTEM_MUTEX_IDENTIFIER,
false), superUser);
+ verifyAllowed(revokePermissions(user,
SYSTEM_CDC_STREAM_IDENTIFIER, false), superUser);
} else {
verifyAllowed(revokePermissions(user,
PHOENIX_SYSTEM_TABLES_IDENTIFIERS, false), superUser);
}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamIT.java
index b4d9ba4642..c515425ecd 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamIT.java
@@ -34,6 +34,8 @@ import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.ManualEnvironmentEdge;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.TestUtil;
@@ -64,7 +66,6 @@ import static
org.apache.phoenix.query.QueryConstants.CDC_PRE_IMAGE;
import static org.apache.phoenix.query.QueryConstants.CDC_UPSERT_EVENT_TYPE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@Category(NeedsOwnMiniClusterTest.class)
@@ -208,6 +209,41 @@ public class CDCStreamIT extends CDCBaseIT {
Assert.assertTrue(rs.next());
}
+ @Test
+ public void testCDCStreamTTL() throws Exception {
+ Connection conn = newConnection();
+ String tableName = generateUniqueName();
+ createTableAndEnableCDC(conn, tableName, true);
+ TestUtil.splitTable(conn, tableName, Bytes.toBytes("m"));
+ String sql = "SELECT PARTITION_END_TIME FROM SYSTEM.CDC_STREAM WHERE
TABLE_NAME='" + tableName + "'";
+ ResultSet rs = conn.createStatement().executeQuery(sql);
+ int count = 0;
+ while (rs.next()) {
+ count++;
+ }
+ Assert.assertEquals(3, count);
+ try {
+ ManualEnvironmentEdge injectEdge = new ManualEnvironmentEdge();
+ long t = System.currentTimeMillis() +
+
QueryServicesOptions.DEFAULT_PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS;
+ t = (t / 1000) * 1000;
+ EnvironmentEdgeManager.injectEdge(injectEdge);
+ injectEdge.setValue(t);
+ rs = conn.createStatement().executeQuery(sql);
+ int newCount = 0;
+ while (rs.next()) {
+ // parent partition row with non-zero end time should have
expired
+ if (rs.getLong(1) > 0) {
+ Assert.fail("Closed partition should have expired after
TTL.");
+ }
+ newCount++;
+ }
+ Assert.assertEquals(2, newCount);
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ }
+
/**
* Split the only region of the table with empty start key and empty end
key.
*/
