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 3a9b391eea PHOENIX-7664 Remove EmptyColumnOnlyFilter and
FirstKeyOnlyFilter for CDC scanners (#2221)
3a9b391eea is described below
commit 3a9b391eea0a467200436703365283c8dbc8de3b
Author: Viraj Jasani <[email protected]>
AuthorDate: Mon Jul 14 10:19:43 2025 -0700
PHOENIX-7664 Remove EmptyColumnOnlyFilter and FirstKeyOnlyFilter for CDC
scanners (#2221)
---
.../java/org/apache/phoenix/util/ScanUtil.java | 47 ++++++--
.../coprocessor/BaseScannerRegionObserver.java | 2 +-
.../java/org/apache/phoenix/end2end/Bson3IT.java | 76 ++++++++++--
.../phoenix/schema/ConditionalTTLExpressionIT.java | 134 +++++++++++++--------
4 files changed, 186 insertions(+), 73 deletions(-)
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 86710eb51b..33724affec 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
@@ -25,13 +25,10 @@ import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverCons
import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.SCAN_STOP_ROW_SUFFIX;
import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.CDC_DATA_TABLE_DEF;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DEFAULT_TTL;
-import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TTL_DEFINED_IN_TABLE_DESCRIPTOR;
import static
org.apache.phoenix.query.QueryConstants.ENCODED_EMPTY_COLUMN_NAME;
import static
org.apache.phoenix.query.QueryServices.USE_STATS_FOR_PARALLELIZATION;
-import static
org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_TABLE_TTL_ENABLED;
import static
org.apache.phoenix.query.QueryServicesOptions.DEFAULT_USE_STATS_FOR_PARALLELIZATION;
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.types.PDataType.TRUE_BYTES;
import static org.apache.phoenix.util.ByteUtil.EMPTY_BYTE_ARRAY;
@@ -44,7 +41,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
-import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
@@ -85,7 +81,6 @@ import
org.apache.phoenix.filter.EncodedQualifiersColumnProjectionFilter;
import org.apache.phoenix.filter.MultiEncodedCQKeyValueComparisonFilter;
import org.apache.phoenix.filter.PagingFilter;
import org.apache.phoenix.filter.SkipScanFilter;
-import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.VersionUtil;
import org.apache.phoenix.index.CDCTableInfo;
@@ -97,9 +92,7 @@ import org.apache.phoenix.query.KeyRange.Bound;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.CompiledConditionalTTLExpression;
import org.apache.phoenix.schema.CompiledTTLExpression;
-import org.apache.phoenix.schema.ConditionalTTLExpression;
import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PName;
@@ -110,7 +103,6 @@ import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeySchema;
import org.apache.phoenix.schema.SortOrder;
-import org.apache.phoenix.schema.TTLExpression;
import org.apache.phoenix.schema.LiteralTTLExpression;
import org.apache.phoenix.schema.TTLExpressionFactory;
import org.apache.phoenix.schema.TableNotFoundException;
@@ -659,6 +651,45 @@ public class ScanUtil {
return false;
}
+ /**
+ * Adjusts the scan filter for CDC scans.
+ * Remove EmptyColumnOnlyFilter and FirstKeyOnlyFilter if present.
+ *
+ * @param scan the scan to adjust.
+ */
+ public static void adjustScanFilterForCDC(Scan scan) {
+ Filter originalFilter = scan.getFilter();
+
+ if (originalFilter instanceof FilterList) {
+ FilterList filterList = (FilterList) originalFilter;
+ List<Filter> filters = filterList.getFilters();
+ List<Filter> newFilters = new ArrayList<>();
+
+ boolean filtersRemoved = false;
+ for (Filter filter : filters) {
+ if (!(filter instanceof EmptyColumnOnlyFilter)
+ && !(filter instanceof FirstKeyOnlyFilter)) {
+ newFilters.add(filter);
+ } else {
+ filtersRemoved = true;
+ }
+ }
+
+ if (filtersRemoved) {
+ if (newFilters.isEmpty()) {
+ scan.setFilter(null);
+ } else if (newFilters.size() == 1) {
+ scan.setFilter(newFilters.get(0));
+ } else {
+ FilterList newFilterList = new
FilterList(filterList.getOperator(), newFilters);
+ scan.setFilter(newFilterList);
+ }
+ }
+ } else {
+ adjustScanFilterForGlobalIndexRegionScanner(scan);
+ }
+ }
+
public static interface BytesComparator {
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int
l2);
};
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index bb46a10cb2..b9eae0ecfb 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -160,7 +160,7 @@ abstract public class BaseScannerRegionObserver implements
RegionObserver {
byte[] cdcScan = scan.getAttribute(CDC_DATA_TABLE_DEF);
if (cdcScan != null) {
CDCUtil.setupScanForCDC(scan);
- ScanUtil.adjustScanFilterForGlobalIndexRegionScanner(scan);
+ ScanUtil.adjustScanFilterForCDC(scan);
}
}
if (isRegionObserverFor(scan)) {
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson3IT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson3IT.java
index fb04899205..841eebb913 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson3IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson3IT.java
@@ -2191,19 +2191,51 @@ public class Bson3IT extends ParallelStatsDisabledIT {
}
// Verify TTL delete events for all rows
- assertEquals("Should have TTL delete events for all 3 rows", 3,
ttlDeleteEvents.size());
-
- // Verify pre-image consistency for each row
- for (String pk : Arrays.asList("pk0001", "pk1010", "pk1011")) {
- Map<String, Object> ttlPreImage = ttlDeleteEvents.get(pk);
- assertNotNull("Should have TTL delete event for " + pk, ttlPreImage);
- Map<String, Object> lastPostImage = lastPostImages.get(pk);
- assertNotNull("TTL pre-image should not be null for " + pk,
ttlPreImage);
- assertNotNull("Last post-image should not be null for " + pk,
lastPostImage);
- assertEquals("TTL delete pre-image should match last post-image for "
+ pk,
- lastPostImage, ttlPreImage);
+ verifyFinalPreImagesWithLastPostImages(ttlDeleteEvents, lastPostImages);
+
+ ttlDeleteQuery = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM " +
cdcName +
+ " WHERE PHOENIX_ROW_TIMESTAMP() > ? LIMIT 1";
+ ttlDeleteEvents = new HashMap<>();
+
+ try (PreparedStatement pst = conn.prepareStatement(ttlDeleteQuery)) {
+ pst.setTimestamp(1, beforeTTLTimestamp);
+ try (ResultSet ttlRs = pst.executeQuery()) {
+ while (ttlRs.next()) {
+ String pk = ttlRs.getString(2);
+ String cdcVal = ttlRs.getString(3);
+ Map<String, Object> cdcEvent = OBJECT_MAPPER.readValue(cdcVal,
HashMap.class);
+
+ // Only process TTL delete events
+ if
(CDC_TTL_DELETE_EVENT_TYPE.equals(cdcEvent.get(CDC_EVENT_TYPE))) {
+ ttlDeleteEvents.put(pk, (Map<String, Object>)
cdcEvent.get(CDC_PRE_IMAGE));
+ }
+ }
+ }
+ }
+
+ assertEquals("Should have TTL delete events for initial 1 row", 1,
ttlDeleteEvents.size());
+
+ ttlDeleteQuery = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM " +
cdcName +
+ " WHERE PHOENIX_ROW_TIMESTAMP() > ? LIMIT 3 OFFSET 1";
+
+ try (PreparedStatement pst = conn.prepareStatement(ttlDeleteQuery)) {
+ pst.setTimestamp(1, beforeTTLTimestamp);
+ try (ResultSet ttlRs = pst.executeQuery()) {
+ while (ttlRs.next()) {
+ String pk = ttlRs.getString(2);
+ String cdcVal = ttlRs.getString(3);
+ Map<String, Object> cdcEvent = OBJECT_MAPPER.readValue(cdcVal,
HashMap.class);
+
+ // Only process TTL delete events
+ if
(CDC_TTL_DELETE_EVENT_TYPE.equals(cdcEvent.get(CDC_EVENT_TYPE))) {
+ ttlDeleteEvents.put(pk, (Map<String, Object>)
cdcEvent.get(CDC_PRE_IMAGE));
+ }
+ }
+ }
}
+ verifyFinalPreImagesWithLastPostImages(ttlDeleteEvents, lastPostImages);
+
ttlDeleteQuery = "SELECT /*+ CDC_INCLUDE(POST) */ * FROM " + cdcName +
" WHERE PHOENIX_ROW_TIMESTAMP() > ?";
@@ -2220,11 +2252,29 @@ public class Bson3IT extends ParallelStatsDisabledIT {
}
}
}
- assertEquals("Total num of TTL_DELETE events without pre-image should be
3 but it is " +
- nonePreImages, 3, nonePreImages);
+ assertEquals("Total num of TTL_DELETE events without pre-image should be
3 but it is "
+ + nonePreImages, 3, nonePreImages);
} finally {
EnvironmentEdgeManager.reset();
}
}
+ private static void verifyFinalPreImagesWithLastPostImages(
+ Map<String, Map<String, Object>> ttlDeleteEvents,
+ Map<String, Map<String, Object>> lastPostImages) {
+ // Verify TTL delete events for all rows
+ assertEquals("Should have TTL delete events for all 3 rows", 3,
ttlDeleteEvents.size());
+
+ // Verify pre-image consistency for each row
+ for (String pk : Arrays.asList("pk0001", "pk1010", "pk1011")) {
+ Map<String, Object> ttlPreImage = ttlDeleteEvents.get(pk);
+ assertNotNull("Should have TTL delete event for " + pk, ttlPreImage);
+ Map<String, Object> lastPostImage = lastPostImages.get(pk);
+ assertNotNull("TTL pre-image should not be null for " + pk, ttlPreImage);
+ assertNotNull("Last post-image should not be null for " + pk,
lastPostImage);
+ assertEquals("TTL delete pre-image should match last post-image for " +
pk,
+ lastPostImage, ttlPreImage);
+ }
+ }
+
}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/schema/ConditionalTTLExpressionIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/schema/ConditionalTTLExpressionIT.java
index 6815d52ebf..74c02f58d6 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/schema/ConditionalTTLExpressionIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/schema/ConditionalTTLExpressionIT.java
@@ -55,6 +55,7 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.util.Bytes;
@@ -450,31 +451,20 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
assertEquals("Table should contain all inserted rows", 5, actual);
// Query initial CDC events (inserts)
- String cdcQuery = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM " +
- PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "." +
cdcName;
+ String cdcQuery1 = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM "
+ + PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "."
+ cdcName;
- ResultSet resultSet =
conn.createStatement().executeQuery(cdcQuery);
- List<Map<String, Object>> postImageList = new ArrayList<>();
- while (resultSet.next()) {
- String cdcVal = resultSet.getString(4);
- Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
+ String cdcQuery2 = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM "
+ + PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "."
+ cdcName
+ + " LIMIT 7";
- // Validate insert events have no pre-image but have post-image
- Map<String, Object> preImage =
- (Map<String, Object>)
map.get(QueryConstants.CDC_PRE_IMAGE);
- assertTrue("Insert events should have empty pre-image",
preImage.isEmpty());
-
- Map<String, Object> postImage =
- (Map<String, Object>)
map.get(QueryConstants.CDC_POST_IMAGE);
- assertFalse("Insert events should have non-empty post-image",
postImage.isEmpty());
- postImageList.add(postImage);
+ String cdcQuery3 = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM "
+ + PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "."
+ cdcName
+ + " LIMIT 10 OFFSET 2";
- assertEquals("Initial events should be UPSERT type",
- QueryConstants.CDC_UPSERT_EVENT_TYPE,
- map.get(QueryConstants.CDC_EVENT_TYPE));
- }
- assertEquals("Post image list size should be 5 but it is " +
postImageList.size(), 5,
- postImageList.size());
+ List<Map<String, Object>> postImageList1 = getPostImageList(conn,
cdcQuery1, 5);
+ List<Map<String, Object>> postImageList2 = getPostImageList(conn,
cdcQuery2, 5);
+ List<Map<String, Object>> postImageList3 = getPostImageList(conn,
cdcQuery3, 3);
// TTL Expiration - Advance time to trigger TTL expiration
injectEdge.incrementValue(ttl);
@@ -484,32 +474,13 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
actual = TestUtil.getRowCount(conn, tableName, true);
assertEquals("All rows should be expired after TTL", 0, actual);
+ cdcQuery2 = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM "
+ + PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "."
+ cdcName
+ + " LIMIT 11";
// TTL CDC Events - Validate TTL_DELETE events are generated
- resultSet = conn.createStatement().executeQuery(cdcQuery);
- int i = 0;
- while (resultSet.next()) {
- String cdcVal = resultSet.getString(4);
- Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
-
- // Validate TTL delete events
- assertEquals("TTL expired rows should generate TTL_DELETE
events",
- QueryConstants.CDC_TTL_DELETE_EVENT_TYPE,
- map.get(QueryConstants.CDC_EVENT_TYPE));
-
- Map<String, Object> preImage =
- (Map<String, Object>)
map.get(QueryConstants.CDC_PRE_IMAGE);
- assertFalse("TTL_DELETE events should have non-empty
pre-image",
- preImage.isEmpty());
-
- assertNull("TTL_DELETE events should have empty post-image",
- map.get(QueryConstants.CDC_POST_IMAGE));
-
- // TTL delete pre-image should match previous upsert post-image
- assertEquals("TTL_DELETE pre-image should match original
insert post-image",
- postImageList.get(i), preImage);
- i++;
- }
- assertEquals("Num of TTL_DELETE events verified should be 5 but it
is " + i, 5, i);
+ compareTtlPreImagesWithLastPostImages(conn, cdcQuery1,
postImageList1, 5);
+ compareTtlPreImagesWithLastPostImages(conn, cdcQuery2,
postImageList2, 5);
+ compareTtlPreImagesWithLastPostImages(conn, cdcQuery3,
postImageList3, 3);
// Update an expired row to bring it back
injectEdge.incrementValue(1);
@@ -542,13 +513,13 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
validateTable(conn, tableName, expectedCellCount,
dataRowPosToKey.values());
// Query CDC events
- cdcQuery = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM " +
+ String cdcQuery = "SELECT /*+ CDC_INCLUDE(PRE, POST) */ * FROM " +
PhoenixTestBuilder.DDLDefaults.DEFAULT_SCHEMA_NAME + "." +
cdcName
+ " WHERE PHOENIX_ROW_TIMESTAMP() >= ?";
PreparedStatement ps = conn.prepareStatement(cdcQuery);
ps.setTimestamp(1, new Timestamp(currentTime));
- resultSet = ps.executeQuery();
- postImageList = new ArrayList<>();
+ ResultSet resultSet = ps.executeQuery();
+ List<Map<String, Object>> postImageList = new ArrayList<>();
while (resultSet.next()) {
String cdcVal = resultSet.getString(4);
Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
@@ -585,7 +556,7 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
ps = conn.prepareStatement(cdcQuery);
ps.setTimestamp(1, new Timestamp(currentTime));
resultSet = ps.executeQuery();
- i = 0;
+ int i = 0;
while (resultSet.next()) {
String cdcVal = resultSet.getString(4);
Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
@@ -610,6 +581,67 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
}
}
+ private static void compareTtlPreImagesWithLastPostImages(Connection conn,
String cdcQuery,
+ List<Map<String,
Object>> postImageList,
+ int
expectedCount)
+ throws SQLException, JsonProcessingException {
+ ResultSet resultSet = conn.createStatement().executeQuery(cdcQuery);
+ int i = 0;
+ while (resultSet.next()) {
+ String cdcVal = resultSet.getString(4);
+ Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
+
+ // Validate TTL delete events
+ assertEquals("TTL expired rows should generate TTL_DELETE events",
+ QueryConstants.CDC_TTL_DELETE_EVENT_TYPE,
+ map.get(QueryConstants.CDC_EVENT_TYPE));
+
+ Map<String, Object> preImage =
+ (Map<String, Object>)
map.get(QueryConstants.CDC_PRE_IMAGE);
+ assertFalse("TTL_DELETE events should have non-empty pre-image",
+ preImage.isEmpty());
+
+ assertNull("TTL_DELETE events should have empty post-image",
+ map.get(QueryConstants.CDC_POST_IMAGE));
+
+ // TTL delete pre-image should match previous upsert post-image
+ assertEquals("TTL_DELETE pre-image should match original insert
post-image",
+ postImageList.get(i), preImage);
+ i++;
+ }
+ assertEquals(
+ "Num of TTL_DELETE events verified should be " + expectedCount
+ " but it is " + i,
+ expectedCount, i);
+ }
+
+ private static List<Map<String, Object>> getPostImageList(Connection conn,
String cdcQuery,
+ int
expectedCount)
+ throws SQLException, JsonProcessingException {
+ ResultSet resultSet = conn.createStatement().executeQuery(cdcQuery);
+ List<Map<String, Object>> postImageList = new ArrayList<>();
+ while (resultSet.next()) {
+ String cdcVal = resultSet.getString(4);
+ Map<String, Object> map = OBJECT_MAPPER.readValue(cdcVal,
Map.class);
+
+ // Validate insert events have no pre-image but have post-image
+ Map<String, Object> preImage =
+ (Map<String, Object>)
map.get(QueryConstants.CDC_PRE_IMAGE);
+ assertTrue("Insert events should have empty pre-image",
preImage.isEmpty());
+
+ Map<String, Object> postImage =
+ (Map<String, Object>)
map.get(QueryConstants.CDC_POST_IMAGE);
+ assertFalse("Insert events should have non-empty post-image",
postImage.isEmpty());
+ postImageList.add(postImage);
+
+ assertEquals("Initial events should be UPSERT type",
+ QueryConstants.CDC_UPSERT_EVENT_TYPE,
+ map.get(QueryConstants.CDC_EVENT_TYPE));
+ }
+ assertEquals("Post image list size should be " + expectedCount + " but
it is "
+ + postImageList.size(), expectedCount, postImageList.size());
+ return postImageList;
+ }
+
@Test
public void testDeleteMarkers() throws Exception {
String ttlCol = "VAL5";
