Support for raw_row_size()

Author: Kadir Ozdemir

phoenix-core-client/src/main/java/org/apache/phoenix/compile/StatementContext.java

@@ -90,6 +90,7 @@ public class StatementContext {
   private boolean totalSegmentsFunction = false;
   private Integer totalSegmentsValue;
   private boolean hasRowSizeFunction = false;
+  private boolean hasRawRowSizeFunction = false;
 
   public StatementContext(PhoenixStatement statement) {
     this(statement, new Scan());
@@ -123,6 +124,7 @@ public StatementContext(StatementContext context) {
     this.totalSegmentsFunction = context.totalSegmentsFunction;
     this.totalSegmentsValue = context.totalSegmentsValue;
     this.hasRowSizeFunction = context.hasRowSizeFunction;
+    this.hasRawRowSizeFunction = context.hasRawRowSizeFunction;
   }
 
   /**
@@ -454,10 +456,18 @@ public boolean hasRowSizeFunction() {
     return hasRowSizeFunction;
   }
 
+  public boolean hasRawRowSizeFunction() {
+    return hasRawRowSizeFunction;
+  }
+
   public void setHasRowSizeFunctionFunction(boolean hasRowSizeFunction) {
     this.hasRowSizeFunction = hasRowSizeFunction;
   }
 
+  public void setHasRawRowSizeFunctionFunction(boolean hasRawRowSizeFunction) {
+    this.hasRawRowSizeFunction = hasRawRowSizeFunction;
+  }
+
   public Integer getTotalSegmentsValue() {
     return totalSegmentsValue;
   }

phoenix-core-client/src/main/java/org/apache/phoenix/compile/WhereCompiler.java

@@ -543,7 +543,7 @@ private static void setScanFilter(StatementContext context, FilterableStatement
       }
     } else if (
       whereClause != null && !ExpressionUtil.evaluatesToTrue(whereClause)
-        && !context.hasRowSizeFunction()
+        && !context.hasRowSizeFunction() && !context.hasRawRowSizeFunction()
     ) {
       Filter filter = null;
       final Counter counter = new Counter();
@@ -604,7 +604,13 @@ public Void visit(KeyValueColumnExpression expression) {
       }
       scan.setFilter(filter);
     } else if (whereClause != null && !ExpressionUtil.evaluatesToTrue(whereClause)) {
-      scan.setFilter(new RowLevelFilter(whereClause));
+      if (context.hasRawRowSizeFunction()) {
+        scan.setFilter(new RowLevelFilter(whereClause, true));
+        scan.setRaw(true);
+        scan.readAllVersions();
+      } else {
+        scan.setFilter(new RowLevelFilter(whereClause, false));
+      }
     }
 
     ScanRanges scanRanges = context.getScanRanges();

phoenix-core-client/src/main/java/org/apache/phoenix/expression/ExpressionType.java

@@ -37,6 +37,7 @@ public enum ExpressionType {
   ReverseFunction(ReverseFunction.class),
   RowKey(RowKeyColumnExpression.class),
   RowSizeFunction(RowSizeFunction.class),
+  RawRowSizeFunction(RawRowSizeFunction.class),
   KeyValue(KeyValueColumnExpression.class),
   LiteralValue(LiteralExpression.class),
   RoundFunction(RoundFunction.class),

phoenix-core-client/src/main/java/org/apache/phoenix/expression/function/RawRowSizeFunction.java

@@ -0,0 +1,41 @@
+/*
+ * 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.expression.function;
+
+import java.sql.SQLException;
+import java.util.List;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.FunctionParseNode.BuiltInFunction;
+import org.apache.phoenix.parse.RawRowSizeParseNode;
+
+/**
+ * Function to return the total size of the all HBase cells versions and delete markers for a given
+ * row
+ */
+@BuiltInFunction(name = RawRowSizeFunction.NAME, nodeClass = RawRowSizeParseNode.class, args = {})
+public class RawRowSizeFunction extends RowSizeFunction {
+
+  public static final String NAME = "RAW_ROW_SIZE";
+
+  public RawRowSizeFunction() {
+  }
+
+  public RawRowSizeFunction(List<Expression> children) throws SQLException {
+    super(children);
+  }
+}

phoenix-core-client/src/main/java/org/apache/phoenix/filter/RowLevelFilter.java

@@ -32,13 +32,15 @@
  * Filter used when expressions reference to the entire row
  */
 public class RowLevelFilter extends BooleanExpressionFilter {
+  private boolean allVersions = false;
   private boolean keepRow = false;
 
   public RowLevelFilter() {
   }
 
-  public RowLevelFilter(Expression expression) {
+  public RowLevelFilter(Expression expression, boolean allVersions) {
     super(expression);
+    this.allVersions = allVersions;
   }
 
   @Override
@@ -54,7 +56,7 @@ public ReturnCode filterKeyValue(Cell v) {
 
   @Override
   public ReturnCode filterCell(Cell v) {
-    return ReturnCode.INCLUDE_AND_NEXT_COL;
+    return allVersions ? ReturnCode.INCLUDE : ReturnCode.INCLUDE_AND_NEXT_COL;
   }
 
   @Override
@@ -72,11 +74,13 @@ public boolean filterRow() {
   @Override
   public void readFields(DataInput input) throws IOException {
     super.readFields(input);
+    allVersions = input.readBoolean();
   }
 
   @Override
   public void write(DataOutput output) throws IOException {
     super.write(output);
+    output.writeBoolean(allVersions);
   }
 
   public static RowLevelFilter parseFrom(final byte[] pbBytes) throws DeserializationException {

phoenix-core-client/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java

@@ -285,7 +285,7 @@ private static void initializeScan(QueryPlan plan, Integer perScanLimit, Integer
       }
       // Add FirstKeyOnlyFilter or EmptyColumnOnlyFilter if there are no references
       // to key value columns. We use FirstKeyOnlyFilter when possible
-      if (keyOnlyFilter && !context.hasRowSizeFunction()) {
+      if (keyOnlyFilter && !context.hasRowSizeFunction() && !context.hasRawRowSizeFunction()) {
         byte[] ecf = SchemaUtil.getEmptyColumnFamily(table);
         byte[] ecq = table.getEncodingScheme() == NON_ENCODED_QUALIFIERS
           ? QueryConstants.EMPTY_COLUMN_BYTES
@@ -361,17 +361,20 @@ private static void initializeScan(QueryPlan plan, Integer perScanLimit, Integer
       if (optimizeProjection) {
         optimizeProjection(context, scan, table, statement);
       }
-      if (context.hasRowSizeFunction()) {
+      if (context.hasRowSizeFunction() || context.hasRawRowSizeFunction()) {
         scan.getFamilyMap().clear();
-        ScanUtil.removePageFilter(scan);
+        if (context.hasRawRowSizeFunction()) {
+          scan.setRaw(true);
+          scan.readAllVersions();
+        }
       }
     }
   }
 
   private static void setQualifierRanges(boolean keyOnlyFilter, PTable table, Scan scan,
     StatementContext context) throws SQLException {
     if (
-      !context.hasRowSizeFunction()
+      !context.hasRowSizeFunction() && !context.hasRawRowSizeFunction()
         && EncodedColumnsUtil.useEncodedQualifierListOptimization(table, scan, context)
     ) {
       Pair<Integer, Integer> minMaxQualifiers = new Pair<>();

phoenix-core-client/src/main/java/org/apache/phoenix/parse/RawRowSizeParseNode.java

@@ -0,0 +1,43 @@
+/*
+ * 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.parse;
+
+import java.sql.SQLException;
+import java.util.List;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.function.FunctionExpression;
+import org.apache.phoenix.expression.function.RawRowSizeFunction;
+
+public class RawRowSizeParseNode extends RowSizeParseNode {
+
+  RawRowSizeParseNode(String name, List<ParseNode> children, BuiltInFunctionInfo info) {
+    super(name, children, info);
+  }
+
+  @Override
+  public FunctionExpression create(List<Expression> children, StatementContext context)
+    throws SQLException {
+    // It does not take any parameters.
+    if (children.size() != 0) {
+      throw new IllegalArgumentException("RawRowSizeFunction does not take any parameters");
+    }
+    context.setHasRawRowSizeFunctionFunction(true);
+    return new RawRowSizeFunction(children);
+  }
+}

phoenix-core-client/src/main/java/org/apache/phoenix/util/EncodedColumnsUtil.java

@@ -141,8 +141,8 @@ public static boolean useEncodedQualifierListOptimization(PTable table, Scan sca
      * on the scan to use this optimization. Disabling this optimization for tables with more than
      * one column family. See PHOENIX-3890.
      */
-    return !scan.isRaw() && !context.hasRowSizeFunction() && table.getColumnFamilies().size() == 1
-      && table.getImmutableStorageScheme() != null
+    return !scan.isRaw() && !context.hasRowSizeFunction() && !context.hasRawRowSizeFunction()
+      && table.getColumnFamilies().size() == 1 && table.getImmutableStorageScheme() != null
       && table.getImmutableStorageScheme() == ImmutableStorageScheme.ONE_CELL_PER_COLUMN
       && usesEncodedColumnNames(table) && !table.isTransactional()
       && !ScanUtil.hasDynamicColumns(table);

phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseAggregateIT.java

@@ -267,7 +267,7 @@ public void testCount() throws Exception {
   }
 
   @Test
-  public void testRowSizeFunction() throws Exception {
+  public void testRowSizeFunctions() throws Exception {
     Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
     Connection conn = DriverManager.getConnection(getUrl(), props);
     String tableName = generateUniqueName();
@@ -352,6 +352,64 @@ public void testRowSizeFunction() throws Exception {
     } catch (SQLException e) {
       // expected
     }
+
+    // Make sure row size functions works with multi-tenant tables
+    tableName = generateUniqueName();
+    String sql = "CREATE TABLE " + tableName
+      + " (ORGANIZATION_ID VARCHAR NOT NULL, CONTAINER_ID VARCHAR, ENTITY_ID INTEGER,"
+      + " CONSTRAINT PK PRIMARY KEY (ORGANIZATION_ID, CONTAINER_ID)) MULTI_TENANT = TRUE";
+    conn.createStatement().execute(sql);
+
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('a','1', 11)");
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('b','2', 22)");
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('c','3', 33)");
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('d','1', 44)");
+    conn.commit();
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('a','1', 12)");
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('a','2', 11)");
+    conn.commit();
+    conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('c','9', 11)");
+    conn.commit();
+    conn.createStatement().execute("DELETE FROM " + tableName + " WHERE organization_id='d'");
+    conn.commit();
+
+    queryBuilder = new QueryBuilder().setSelectExpression("ORGANIZATION_ID, sum(row_size())")
+      .setFullTableName(tableName).setGroupByClause("ORGANIZATION_ID");
+    rs = executeQuery(conn, queryBuilder);
+    assertTrue(rs.next());
+    assertEquals("a", rs.getString(1));
+    assertEquals(118, rs.getLong(2));
+    assertTrue(rs.next());
+    assertEquals("b", rs.getString(1));
+    assertEquals(59, rs.getLong(2));
+    assertTrue(rs.next());
+    assertEquals("c", rs.getString(1));
+    assertEquals(118, rs.getLong(2));
+    assertFalse(rs.next());
+
+    // Make sure raw_row_size() computation includes all cell versions and delete markers
+    queryBuilder = new QueryBuilder().setSelectExpression("ORGANIZATION_ID, sum(raw_row_size())")
+      .setFullTableName(tableName).setGroupByClause("ORGANIZATION_ID");
+    rs = executeQuery(conn, queryBuilder);
+
+    assertTrue(rs.next());
+    assertEquals("a", rs.getString(1));
+    // There are 3 row versions, each version is 59 bytes and so 3 * 59 = 177
+    assertEquals(177, rs.getLong(2));
+    assertTrue(rs.next());
+    assertEquals("b", rs.getString(1));
+    // There is one row version and one version is 59 bytes
+    assertEquals(59, rs.getLong(2));
+    assertTrue(rs.next());
+    assertEquals("c", rs.getString(1));
+    // There are two versions, 2 * 59 = 118
+    assertEquals(118, rs.getLong(2));
+    assertTrue(rs.next());
+    assertEquals("d", rs.getString(1));
+    // One row version (59 bytes) and plus delete family marker
+    assertEquals(83, rs.getLong(2));
+    assertFalse(rs.next());
+
     conn.close();
   }