This is an automated email from the ASF dual-hosted git repository.
tkhurana 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 6ac1c4bc89 PHOENIX-7170 Conditional TTL (addendum)
6ac1c4bc89 is described below
commit 6ac1c4bc89cd99beb30d2e109694bd67bc75b39f
Author: tkhurana <[email protected]>
AuthorDate: Mon Apr 14 16:35:14 2025 -0700
PHOENIX-7170 Conditional TTL (addendum)
---
.../schema/CompiledConditionalTTLExpression.java | 98 ++++++++-----
.../phoenix/schema/ConditionalTTLExpressionIT.java | 19 ++-
.../schema/ConditionalTTLExpressionTest.java | 158 +++++++++++++++++++++
3 files changed, 227 insertions(+), 48 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/CompiledConditionalTTLExpression.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/CompiledConditionalTTLExpression.java
index 334ddd662e..2219bcb75b 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/CompiledConditionalTTLExpression.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/CompiledConditionalTTLExpression.java
@@ -29,7 +29,6 @@ import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Set;
@@ -38,7 +37,6 @@ import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.ByteStringer;
-import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos;
@@ -49,7 +47,6 @@ import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.types.PBoolean;
import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,6 +60,10 @@ public class CompiledConditionalTTLExpression implements
CompiledTTLExpression {
private final Expression compiledExpr;
// columns referenced in the ttl expression to be added to scan
private final Set<ColumnReference> conditionExprColumns;
+ // used to determine the latest version of a row in case of raw scan
+ List<Cell> latestRowVersion = new ArrayList<>();
+ // record DeleteFamilyVersion cells in a row
+ List<Cell> deleteFamilyVersionCells = new ArrayList<>();
public CompiledConditionalTTLExpression(String ttlExpr,
Expression compiledExpression,
@@ -108,48 +109,69 @@ public class CompiledConditionalTTLExpression implements
CompiledTTLExpression {
* the oldest. The method leverages this ordering and groups the cells
into their columns
* based on the pair of family name and column qualifier.
*/
- private List<Cell> getLatestRowVersion(List<Cell> result) {
- List<Cell> latestRowVersion = new ArrayList<>();
- Cell currentColumnCell = null;
- Cell maxDeleteFamily = null;
- Cell maxDeleteFamilyVersion = null;
- for (Cell cell : result) {
+ @VisibleForTesting
+ public List<Cell> getLatestRowVersion(List<Cell> result) {
+ latestRowVersion.clear();
+ deleteFamilyVersionCells.clear();
+ Cell maxDeleteFamily = null; // record the DeleteFamily cell with the
highest timestamp
+ int index = 0;
+ while (index < result.size()) {
+ Cell cell = result.get(index);
if (cell.getType() == Cell.Type.DeleteFamily) {
- if (maxDeleteFamily == null) {
- maxDeleteFamily = cell;
- }
- // no need to add the DeleteFamily cell since it can't be part
of
- // an expression
- continue;
+ // record the first DeleteFamily cell
+ maxDeleteFamily = cell;
+ // we can now skip the delete family column
+ index = skipColumn(result, cell, index);
}
- if (cell.getType() == Cell.Type.DeleteFamilyVersion) {
- if (maxDeleteFamilyVersion == null) {
- maxDeleteFamilyVersion = cell;
- }
- // no need to add the DeleteFamilyVersion cell since it can't
be part of
- // an expression
- continue;
+ else if (cell.getType() == Cell.Type.DeleteFamilyVersion) {
+ // record and skip this cell
+ deleteFamilyVersionCells.add(cell);
+ index++;
}
- if (currentColumnCell == null ||
- !CellUtil.matchingColumn(cell, currentColumnCell)) {
- if (maxDeleteFamilyVersion != null &&
- cell.getTimestamp() ==
maxDeleteFamilyVersion.getTimestamp()) {
- // only add the cell if it is not masked by the
DeleteFamilyVersion
- continue;
- }
- // found a new column cell which has the latest timestamp
- currentColumnCell = cell;
- if (maxDeleteFamily != null &&
+ else if (cell.getType() == Cell.Type.DeleteColumn) {
+ // we can skip all the remaining cells of this column
+ index = skipColumn(result, cell, index);
+ } else if (cell.getType() == Cell.Type.Delete) {
+ index++;
+ } else if (cell.getType() == Cell.Type.Put) {
+ // check if this cell is masked by DeleteFamilyVersion
+ if
(deleteFamilyVersionCells.stream().anyMatch(deleteFamilyCell ->
+ cell.getTimestamp() ==
deleteFamilyCell.getTimestamp())) {
+ // skip the cell
+ index++;
+ } else if (maxDeleteFamily != null &&
cell.getTimestamp() <= maxDeleteFamily.getTimestamp())
{
- // only add the cell if it is not masked by the
DeleteFamily
- continue;
+ // this cell is masked by the DeleteFamily
+ index = skipColumn(result, cell, index);
+ } else {
+ latestRowVersion.add(cell);
+ // we have found the latest cell, skip the remaining cells
of this column
+ index = skipColumn(result, cell, index);
}
- latestRowVersion.add(currentColumnCell);
}
}
return latestRowVersion;
}
+ /**
+ * Skip the remaining cells of the current column
+ * @param result The list of cells (input)
+ * @param currentColumnCell The cell indicates the current column (input)
+ * @param index The index of the current cell in result (input)
+ * @return the index of the first cell of the next column
+ */
+ private int skipColumn(List<Cell> result, Cell currentColumnCell, int
index) {
+ int next = index + 1;
+ while (next < result.size()) {
+ Cell cell = result.get(next);
+ if (!CellUtil.matchingColumn(cell, currentColumnCell)) {
+ break;
+ }
+ next++;
+ }
+ return next;
+ }
+
@Override
/**
* @param result row to be evaluated against the conditional ttl
expression. The cells
@@ -184,11 +206,11 @@ public class CompiledConditionalTTLExpression implements
CompiledTTLExpression {
return false;
}
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- List<Cell> latestRowVersion = !isRaw ? result :
getLatestRowVersion(result);
- if (latestRowVersion.isEmpty()) {
+ List<Cell> latestRow = !isRaw ? result : getLatestRowVersion(result);
+ if (latestRow.isEmpty()) {
return false;
}
- MultiKeyValueTuple row = new MultiKeyValueTuple(latestRowVersion);
+ MultiKeyValueTuple row = new MultiKeyValueTuple(latestRow);
if (!compiledExpr.evaluate(row, ptr)) {
LOGGER.info("Expression evaluation failed for expr {}", ttlExpr);
return false;
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 4e3fa275de..4b73197631 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
@@ -748,7 +748,7 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
TestUtil.dumpTable(conn, TableName.valueOf(fullIndexName));
assertEquals(rowCount - 1, actual);
// First read row 0 which is not expired
- String dql = String.format("select VAL2, VAL5 from %s where
VAL1='%s'",
+ String dql = String.format("select VAL2, VAL5 from %s where
VAL1='%s' AND ID2=0",
fullDataTableName, val1_0);
try (ResultSet rs1 = conn.createStatement().executeQuery(dql)) {
PhoenixResultSet prs = rs1.unwrap(PhoenixResultSet.class);
@@ -759,7 +759,7 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
assertFalse(rs1.getBoolean(ttlCol));
}
// row 1 is expired
- dql = String.format("select VAL2, VAL5 from %s where VAL1='%s'",
+ dql = String.format("select VAL2, VAL5 from %s where VAL1='%s' AND
ID2=1",
fullDataTableName, val1_1);
try (ResultSet rs1 = conn.createStatement().executeQuery(dql)) {
PhoenixResultSet prs = rs1.unwrap(PhoenixResultSet.class);
@@ -792,14 +792,13 @@ public class ConditionalTTLExpressionIT extends
ParallelStatsDisabledIT {
IndexToolIT.dumpMRJobCounters(mrJobCounters);
throw e;
}
- // TODO: Broken because of PHOENIX-7574
- //doMajorCompaction(fullIndexName);
- //TestUtil.dumpTable(conn, TableName.valueOf(fullIndexName));
- //actual = TestUtil.getRowCountFromIndex(conn, fullDataTableName,
fullIndexName);
- //assertEquals(rowCount - 1, actual);
- //CellCount expectedCellCount = new CellCount();
- //expectedCellCount.insertRow(indexRowPosToKey.get(0), 2);
- //validateTable(conn, fullIndexName, expectedCellCount,
indexRowPosToKey.values());
+ doMajorCompaction(fullIndexName);
+ TestUtil.dumpTable(conn, TableName.valueOf(fullIndexName));
+ actual = TestUtil.getRowCountFromIndex(conn, fullDataTableName,
fullIndexName);
+ assertEquals(rowCount - 1, actual);
+ CellCount expectedCellCount = new CellCount();
+ expectedCellCount.insertRow(indexRowPosToKey.get(0),
includedColumns.size() + 1);
+ validateTable(conn, fullIndexName, expectedCellCount,
indexRowPosToKey.values());
}
}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/schema/ConditionalTTLExpressionTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/schema/ConditionalTTLExpressionTest.java
index d20dd901c3..fddae2229f 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/schema/ConditionalTTLExpressionTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/schema/ConditionalTTLExpressionTest.java
@@ -29,22 +29,37 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.exception.PhoenixParserException;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
import org.apache.phoenix.query.BaseConnectionlessQueryTest;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.junit.Test;
@@ -639,4 +654,147 @@ public class ConditionalTTLExpressionTest extends
BaseConnectionlessQueryTest {
validateScan(conn, tableName, query, ttl, false,
Lists.newArrayList("col1", "col2"));
}
}
+
+ @Test
+ public void testLatestRowVersion() throws Exception {
+ String ddlTemplate = "create table %s (id varchar not null primary
key, " +
+ "col1 integer, col2 integer, col3 varchar) TTL = '%s'";
+ String tableName = generateUniqueName();
+ String ttl = "col2 > 100 AND col3='expired'";
+ long currentTS = EnvironmentEdgeManager.currentTimeMillis();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ String ddl = String.format(ddlTemplate, tableName,
retainSingleQuotes(ttl));
+ conn.createStatement().execute(ddl);
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ PTable table = pconn.getTable(tableName);
+ int updateCount = 5;
+ String id = "a";
+ ImmutableBytesPtr rowKey = new
ImmutableBytesPtr(Bytes.toBytes(id));
+ List<Cell> updates = new ArrayList<>();
+ for (int i = 0; i < updateCount; ++i) {
+ updates.addAll(generateUpdate(id, i, i*i, "val_" + i,
currentTS + i*10));
+ }
+ // updates [
+ // lastUpdateTS - 40,
+ // lastUpdateTS - 30,
+ // lastUpdateTS - 20,
+ // lastUpdateTS - 10,
+ // lastUpdateTS
+ // ]
+ long lastupdateTS = currentTS + (updateCount - 1)* 10;
+
+ CompiledConditionalTTLExpression ttlExpr =
+ (CompiledConditionalTTLExpression)
table.getCompiledTTLExpression(pconn);
+
+ // case 1: no Deletes just multiple Put mutations
+ List<Cell> row = Lists.newArrayList(updates);
+ row.sort(CellComparator.getInstance());
+ List<Cell> latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(3, latestRowVersion.size()); // 3 columns
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS, cell.getTimestamp());
+ }
+
+ // case 2 insert DeleteFamily at the top
+ row = Lists.newArrayList(updates);
+ KeyValue df = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
null,
+ lastupdateTS + 5, KeyValue.Type.DeleteFamily, null);
+ row.add(df);
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertTrue(latestRowVersion.isEmpty());
+
+ // case 3 insert DeleteFamilyVersion at last TS
+ row = Lists.newArrayList(updates);
+ KeyValue dfv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
null, lastupdateTS,
+ KeyValue.Type.DeleteFamilyVersion, null);
+ row.add(dfv);
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(3, latestRowVersion.size()); // 3 columns
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS - 10, cell.getTimestamp());
+ }
+
+ // case 4 insert DeleteFamilyVersion at last TS, DeleteFamily in
the middle
+ row = Lists.newArrayList(updates);
+ df = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"), null,
+ lastupdateTS - 25, KeyValue.Type.DeleteFamily, null);
+ row.add(df);
+ row.add(dfv);
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(3, latestRowVersion.size()); // 3 columns
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS - 10, cell.getTimestamp());
+ }
+
+ // case 5 multiple DeleteFamilyVersions
+ row = Lists.newArrayList(updates);
+ row.add(dfv);
+ dfv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"), null,
lastupdateTS - 10,
+ KeyValue.Type.DeleteFamilyVersion, null);
+ row.add(dfv);
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(3, latestRowVersion.size()); // 3 columns
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS - 20, cell.getTimestamp());
+ }
+
+ // case 6 DeleteFamily in the middle
+ row = Lists.newArrayList(updates);
+ df = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"), null,
+ lastupdateTS - 5, KeyValue.Type.DeleteFamily, null);
+ row.add(df);
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(3, latestRowVersion.size()); // 3 columns
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS, cell.getTimestamp());
+ }
+
+ // case 7 DeleteColumn
+ row = Lists.newArrayList(updates);
+ row.addAll(generateUpdate(id, 12, null, null, lastupdateTS + 10));
+ row.sort(CellComparator.getInstance());
+ latestRowVersion = ttlExpr.getLatestRowVersion(row);
+ assertEquals(1, latestRowVersion.size()); // 1 column is non null
+ for (Cell cell : latestRowVersion) {
+ assertEquals(lastupdateTS + 10, cell.getTimestamp());
+ }
+ }
+ }
+
+ private List<Cell> generateUpdate(String id, Integer col1, Integer col2,
String col3, long ts) {
+ List<Cell> update = Lists.newArrayList();
+ if (col1 != null) {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col1"),
+ ts, KeyValue.Type.Put, Bytes.toBytes(col1));
+ update.add(kv);
+ } else {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col1"),
+ ts, KeyValue.Type.DeleteColumn, null);
+ update.add(kv);
+ }
+ if (col2 != null) {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col2"),
+ ts, KeyValue.Type.Put, Bytes.toBytes(col2));
+ update.add(kv);
+ } else {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col2"),
+ ts, KeyValue.Type.DeleteColumn, null);
+ update.add(kv);
+ }
+ if (col3 != null) {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col3"),
+ ts, KeyValue.Type.Put, Bytes.toBytes(col3));
+ update.add(kv);
+ } else {
+ KeyValue kv = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("f"),
Bytes.toBytes("col3"),
+ ts, KeyValue.Type.DeleteColumn, null);
+ update.add(kv);
+ }
+ return update;
+ }
}
