shubham-roy commented on code in PR #2614:
URL: https://github.com/apache/phoenix/pull/2614#discussion_r3964722028
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CompactionScanner.java:
##########
@@ -1156,10 +1156,15 @@ public NonPartitionedTableTTLTracker(PhoenixConnection
pConn, PTable pTable, Sto
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 isConditionalTTL = pTable.getTTLExpression() instanceof
ConditionalTTLExpression;
Review Comment:
Done — switched to `pTable.hasConditionalTTL()` and renamed the local to
`conditionalTTL`. The same change is on the master PR #2613. Thanks for the
review!
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamCompactionTTLIT.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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);
Review Comment:
Good call. After the major compaction I now re-query `SYSTEM.CDC_STREAM` via
SQL and assert `total == 2` and `closed == 0`, alongside the existing raw HBase
row-count check (`== 2`) — so the test confirms the read path agrees with the
physical purge, not just the on-disk state. Verified green on the minicluster
IT (`Tests run: 1, Failures: 0`). Applied here and mirrored on the master PR
#2613.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
