This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch 5.3
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/5.3 by this push:
new 9d4ad20b13 PHOENIX-8000 Conditional TTL not enforced at compaction for
SYSTEM tables (#2614)
9d4ad20b13 is described below
commit 9d4ad20b134605d22acae2a86c935b7cf83bb24a
Author: Shubham Roy <[email protected]>
AuthorDate: Thu Sep 10 09:48:10 2026 +0530
PHOENIX-8000 Conditional TTL not enforced at compaction for SYSTEM tables
(#2614)
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../phoenix/coprocessor/CompactionScanner.java | 7 +-
.../phoenix/end2end/CDCStreamCompactionTTLIT.java | 153 +++++++++++++++++++++
2 files changed, 159 insertions(+), 1 deletion(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CompactionScanner.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CompactionScanner.java
index 7e79c00b72..faec302764 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CompactionScanner.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CompactionScanner.java
@@ -1156,10 +1156,15 @@ public class CompactionScanner implements
InternalScanner {
throws IOException {
boolean isSystemTable = pTable.getType() == PTableType.SYSTEM;
+ // A SYSTEM table with a CONDITIONAL TTL (e.g. SYSTEM.CDC_STREAM) must
NOT be forced onto the
+ // column-family descriptor TTL (which is FOREVER for such tables); its
conditional expression
+ // has to be compiled so that expired rows are physically purged at
major compaction, not
+ // merely masked at read time.
+ boolean conditionalTTL = pTable.hasConditionalTTL();
boolean ttlFromDescriptor = false;
try {
if (
- isSystemTable
+ isSystemTable && !conditionalTTL
||
pTable.getTTLExpression().equals(TTL_EXPRESSION_DEFINED_IN_TABLE_DESCRIPTOR)
) {
ColumnFamilyDescriptor cfd = store.getColumnFamilyDescriptor();
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamCompactionTTLIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamCompactionTTLIT.java
new file mode 100644
index 0000000000..87c41ba625
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamCompactionTTLIT.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.end2end;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY;
+import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.util.Map;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessor.PhoenixMasterObserver;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+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;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+/**
+ * Verifies that a major compaction of a SYSTEM table whose TTL is a
CONDITIONAL expression
+ * (SYSTEM.CDC_STREAM) physically purges expired rows, rather than only
masking them at read time.
+ * <p>
+ * SYSTEM.CDC_STREAM is created with a conditional TTL that expires CLOSED
partition rows
+ * (PARTITION_END_TIME IS NOT NULL) once they age past the configured
partition-expiry window. If
+ * the compaction path routes SYSTEM tables unconditionally onto the
column-family-descriptor TTL
+ * (which is FOREVER for this table), the conditional expression is never
compiled, so
+ * postProcessForConditionalTTL never runs and the expired closed-partition
rows survive on disk
+ * forever. This IT drives a major compaction after advancing the clock past
the expiry window and
+ * asserts the closed-partition rows are physically gone.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class CDCStreamCompactionTTLIT extends CDCBaseIT {
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(5);
+ // Conditional-TTL rows are only PHYSICALLY purged at major compaction
once they fall outside
+ // the max-lookback retain window. Disable max-lookback (0) so the expired
closed-partition
+ // rows are eligible for physical removal at compaction, not merely masked
at read time.
+ props.put(PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY, Integer.toString(0));
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION,
Boolean.toString(false));
+ props.put(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB,
Long.toString(Long.MAX_VALUE));
+ props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB,
Long.toString(Long.MAX_VALUE));
+ props.put("hbase.coprocessor.master.classes",
PhoenixMasterObserver.class.getName());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ taskRegionEnvironment =
+
getUtility().getRSForFirstRegionInTable(PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME)
+
.getRegions(PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME).get(0)
+
.getCoprocessorHost().findCoprocessorEnvironment(TaskRegionObserver.class.getName());
+ }
+
+ @Test
+ public void testMajorCompactionPhysicallyPurgesExpiredClosedPartitionRows()
throws Exception {
+ Connection conn = newConnection();
+ String tableName = generateUniqueName();
+ createTableAndEnableCDC(conn, tableName, true);
+ // Splitting yields 3 SYSTEM.CDC_STREAM rows for this table: 1 CLOSED
parent partition
+ // (PARTITION_END_TIME not null) + 2 OPEN child partitions
(PARTITION_END_TIME null).
+ TestUtil.splitTable(conn, tableName, Bytes.toBytes("m"));
+
+ String totalSql = "SELECT COUNT(*) FROM SYSTEM.CDC_STREAM WHERE
TABLE_NAME='" + tableName + "'";
+ // The conditional TTL expires exactly the rows with PARTITION_END_TIME IS
NOT NULL.
+ String closedSql = "SELECT COUNT(*) FROM SYSTEM.CDC_STREAM WHERE
TABLE_NAME='" + tableName
+ + "' AND PARTITION_END_TIME IS NOT NULL";
+
+ long total;
+ long closed;
+ try (ResultSet rs = conn.createStatement().executeQuery(totalSql)) {
+ assertTrue(rs.next());
+ total = rs.getLong(1);
+ }
+ try (ResultSet rs = conn.createStatement().executeQuery(closedSql)) {
+ assertTrue(rs.next());
+ closed = rs.getLong(1);
+ }
+ // A single split of a one-region table yields exactly 3 SYSTEM.CDC_STREAM
rows:
+ // 1 closed parent partition (PARTITION_END_TIME not null) + 2 open child
partitions.
+ assertEquals("split should yield exactly 3 partition rows", 3, total);
+ assertEquals("exactly one closed (split-parent) partition row", 1, closed);
+
+ // Resolve the physical HBase table backing SYSTEM.CDC_STREAM.
+ TableName physicalCdcStreamTable =
SchemaUtil.getPhysicalTableName(SYSTEM_CDC_STREAM_NAME,
+ conn.unwrap(PhoenixConnection.class).getQueryServices().getProps());
+ long rawBefore = TestUtil.getRawRowCount(conn, physicalCdcStreamTable);
+ // Before advancing the clock, every partition row is still present on
disk.
+ assertEquals("baseline: all 3 partition rows present on disk before
compaction", 3, rawBefore);
+
+ try {
+ // Advance the clock past the partition-expiry window so the
closed-partition rows expire.
+ ManualEnvironmentEdge injectEdge = new ManualEnvironmentEdge();
+ long t = System.currentTimeMillis()
+ +
QueryServicesOptions.DEFAULT_PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS +
5000;
+ t = (t / 1000) * 1000;
+ EnvironmentEdgeManager.injectEdge(injectEdge);
+ injectEdge.setValue(t);
+
+ // Major compact SYSTEM.CDC_STREAM. With the conditional TTL compiled,
the expired
+ // closed-partition rows must be physically removed here.
+ TestUtil.doMajorCompaction(conn, SYSTEM_CDC_STREAM_NAME);
+
+ long rawAfter = TestUtil.getRawRowCount(conn, physicalCdcStreamTable);
+ assertEquals("closed partition physically purged; only the 2 open
partitions remain", 2,
+ rawAfter);
+
+ // The SQL (read-path) view must now agree with the physical purge: the
closed parent
+ // partition row is gone and only the 2 open child partitions remain
visible.
+ long totalAfter;
+ long closedAfter;
+ try (ResultSet rs = conn.createStatement().executeQuery(totalSql)) {
+ assertTrue(rs.next());
+ totalAfter = rs.getLong(1);
+ }
+ try (ResultSet rs = conn.createStatement().executeQuery(closedSql)) {
+ assertTrue(rs.next());
+ closedAfter = rs.getLong(1);
+ }
+ assertEquals("only the 2 open partitions remain visible via SQL after
compaction", 2,
+ totalAfter);
+ assertEquals("no closed partition rows remain visible via SQL after
compaction", 0,
+ closedAfter);
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ }
+}
