You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2018/03/09 23:21:54 UTC
[1/4] phoenix git commit: PHOENIX-4640 Don't consider
STATS_UPDATE_FREQ_MS_ATTRIB in TTL for server side cache
Repository: phoenix
Updated Branches:
refs/heads/master 916186e97 -> d2715101e
PHOENIX-4640 Don't consider STATS_UPDATE_FREQ_MS_ATTRIB in TTL for server side cache
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/32a8f3b2
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/32a8f3b2
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/32a8f3b2
Branch: refs/heads/master
Commit: 32a8f3b290d29a1203485cf8aa6649f338277894
Parents: 916186e
Author: James Taylor <jt...@salesforce.com>
Authored: Mon Mar 5 16:27:31 2018 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Mar 9 15:18:44 2018 -0800
----------------------------------------------------------------------
.../src/main/java/org/apache/phoenix/cache/GlobalCache.java | 6 ++----
1 file changed, 2 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/32a8f3b2/phoenix-core/src/main/java/org/apache/phoenix/cache/GlobalCache.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/cache/GlobalCache.java b/phoenix-core/src/main/java/org/apache/phoenix/cache/GlobalCache.java
index 43df5db..ae77174 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/cache/GlobalCache.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/cache/GlobalCache.java
@@ -92,11 +92,9 @@ public class GlobalCache extends TenantCacheImpl {
synchronized(this) {
result = metaDataCache;
if(result == null) {
- long maxTTL = Math.min(config.getLong(
+ long maxTTL = config.getLong(
QueryServices.MAX_SERVER_METADATA_CACHE_TIME_TO_LIVE_MS_ATTRIB,
- QueryServicesOptions.DEFAULT_MAX_SERVER_METADATA_CACHE_TIME_TO_LIVE_MS), config.getLong(
- QueryServices.STATS_UPDATE_FREQ_MS_ATTRIB,
- QueryServicesOptions.DEFAULT_STATS_UPDATE_FREQ_MS));
+ QueryServicesOptions.DEFAULT_MAX_SERVER_METADATA_CACHE_TIME_TO_LIVE_MS);
long maxSize = config.getLong(QueryServices.MAX_SERVER_METADATA_CACHE_SIZE_ATTRIB,
QueryServicesOptions.DEFAULT_MAX_SERVER_METADATA_CACHE_SIZE);
metaDataCache = result = CacheBuilder.newBuilder()
[3/4] phoenix git commit: PHOENIX-1267 Set scan.setSmall(true) when
appropriate (Abhishek Singh Chouhan)
Posted by ja...@apache.org.
PHOENIX-1267 Set scan.setSmall(true) when appropriate (Abhishek Singh Chouhan)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/abfc1ff4
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/abfc1ff4
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/abfc1ff4
Branch: refs/heads/master
Commit: abfc1ff4d37d67eb4666c4c7db24cb22f041768c
Parents: 45f856a
Author: James Taylor <jt...@salesforce.com>
Authored: Thu Mar 8 14:38:28 2018 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Mar 9 15:19:20 2018 -0800
----------------------------------------------------------------------
.../apache/phoenix/execute/BaseQueryPlan.java | 16 +++++++++---
.../org/apache/phoenix/query/QueryServices.java | 1 +
.../phoenix/query/QueryServicesOptions.java | 1 +
.../phoenix/compile/QueryCompilerTest.java | 26 ++++++++++++++++++++
4 files changed, 40 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/abfc1ff4/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
index 0bc606e..b152030 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
@@ -63,6 +63,8 @@ import org.apache.phoenix.parse.HintNode.Hint;
import org.apache.phoenix.parse.ParseNodeFactory;
import org.apache.phoenix.parse.TableName;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.KeyValueSchema;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PName;
@@ -235,12 +237,14 @@ public abstract class BaseQueryPlan implements QueryPlan {
scan = context.getScan();
}
+ ScanRanges scanRanges = context.getScanRanges();
+
/*
* For aggregate queries, we still need to let the AggregationPlan to
* proceed so that we can give proper aggregates even if there are no
* row to be scanned.
*/
- if (context.getScanRanges() == ScanRanges.NOTHING && !getStatement().isAggregate()) {
+ if (scanRanges == ScanRanges.NOTHING && !getStatement().isAggregate()) {
return getWrappedIterator(caches, ResultIterator.EMPTY_ITERATOR);
}
@@ -267,11 +271,15 @@ public abstract class BaseQueryPlan implements QueryPlan {
}
}
- if (statement.getHint().hasHint(Hint.SMALL)) {
+
+ PhoenixConnection connection = context.getConnection();
+ final int smallScanThreshold = connection.getQueryServices().getProps().getInt(QueryServices.SMALL_SCAN_THRESHOLD_ATTRIB,
+ QueryServicesOptions.DEFAULT_SMALL_SCAN_THRESHOLD);
+
+ if (statement.getHint().hasHint(Hint.SMALL) || (scanRanges.isPointLookup() && scanRanges.getPointLookupCount() < smallScanThreshold)) {
scan.setSmall(true);
}
- PhoenixConnection connection = context.getConnection();
// set read consistency
if (table.getType() != PTableType.SYSTEM) {
@@ -280,7 +288,7 @@ public abstract class BaseQueryPlan implements QueryPlan {
// TODO fix this in PHOENIX-2415 Support ROW_TIMESTAMP with transactional tables
if (!table.isTransactional()) {
// Get the time range of row_timestamp column
- TimeRange rowTimestampRange = context.getScanRanges().getRowTimestampRange();
+ TimeRange rowTimestampRange = scanRanges.getRowTimestampRange();
// Get the already existing time range on the scan.
TimeRange scanTimeRange = scan.getTimeRange();
Long scn = connection.getSCN();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/abfc1ff4/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index 0979e36..0b18aaa 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -300,6 +300,7 @@ public interface QueryServices extends SQLCloseable {
// Whether to enable cost-based-decision in the query optimizer
public static final String COST_BASED_OPTIMIZER_ENABLED = "phoenix.costbased.optimizer.enabled";
+ public static final String SMALL_SCAN_THRESHOLD_ATTRIB = "phoenix.query.smallScanThreshold";
/**
* Get executor service used for parallel scans
http://git-wip-us.apache.org/repos/asf/phoenix/blob/abfc1ff4/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 8dadbb0..961ab9f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -327,6 +327,7 @@ public class QueryServicesOptions {
//default update cache frequency
public static final int DEFAULT_UPDATE_CACHE_FREQUENCY = 0;
+ public static final int DEFAULT_SMALL_SCAN_THRESHOLD = 100;
@SuppressWarnings("serial")
public static final Set<String> DEFAULT_QUERY_SERVER_SKIP_WORDS = new HashSet<String>() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/abfc1ff4/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
index 9386bd2..2b5a8cb 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
@@ -4483,6 +4483,32 @@ public class QueryCompilerTest extends BaseConnectionlessQueryTest {
}
@Test
+ public void testSmallScanForPointLookups() throws SQLException {
+ Properties props = PropertiesUtil.deepCopy(new Properties());
+ createTestTable(getUrl(), "CREATE TABLE FOO(\n" +
+ " a VARCHAR NOT NULL,\n" +
+ " b VARCHAR NOT NULL,\n" +
+ " c VARCHAR,\n" +
+ " CONSTRAINT pk PRIMARY KEY (a, b DESC, c)\n" +
+ " )");
+
+ props.put(QueryServices.SMALL_SCAN_THRESHOLD_ATTRIB, "2");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String query = "select * from foo where a = 'a' and b = 'b' and c in ('x','y','z')";
+ PhoenixStatement stmt = conn.createStatement().unwrap(PhoenixStatement.class);
+ QueryPlan plan = stmt.optimizeQuery(query);
+ plan.iterator();
+ //Fail since we have 3 rows in pointLookup
+ assertFalse(plan.getContext().getScan().isSmall());
+ query = "select * from foo where a = 'a' and b = 'b' and c = 'c'";
+ plan = stmt.compileQuery(query);
+ plan.iterator();
+ //Should be small scan, query is for single row pointLookup
+ assertTrue(plan.getContext().getScan().isSmall());
+ }
+ }
+
+ @Test
public void testLocalIndexPruningInSortMergeJoin() throws SQLException {
verifyLocalIndexPruningWithMultipleTables("SELECT /*+ USE_SORT_MERGE_JOIN*/ *\n" +
"FROM T1 JOIN T2 ON T1.A = T2.A\n" +
[2/4] phoenix git commit: PHOENIX-4148 COUNT(DISTINCT(...)) should
have a memory size limit (Lars Hofhansl)
Posted by ja...@apache.org.
PHOENIX-4148 COUNT(DISTINCT(...)) should have a memory size limit (Lars Hofhansl)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/45f856af
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/45f856af
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/45f856af
Branch: refs/heads/master
Commit: 45f856af776a29eaf6ae503d6d3062b277893396
Parents: 32a8f3b
Author: James Taylor <jt...@salesforce.com>
Authored: Tue Mar 6 15:03:36 2018 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Mar 9 15:19:00 2018 -0800
----------------------------------------------------------------------
.../phoenix/end2end/SpillableGroupByIT.java | 74 ++++++++++++++------
.../UngroupedAggregateRegionObserver.java | 13 +++-
.../phoenix/exception/SQLExceptionCode.java | 5 +-
.../expression/aggregator/Aggregators.java | 3 +-
.../aggregator/ClientAggregators.java | 3 +-
.../DistinctValueWithCountServerAggregator.java | 20 ++----
.../aggregator/ServerAggregators.java | 6 +-
.../phoenix/memory/ChildMemoryManager.java | 7 +-
.../phoenix/memory/GlobalMemoryManager.java | 19 +++--
.../java/org/apache/phoenix/util/TestUtil.java | 1 -
10 files changed, 102 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
index dc04b53..3689c4c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
@@ -17,12 +17,12 @@
*/
package org.apache.phoenix.end2end;
-import static org.apache.phoenix.util.TestUtil.GROUPBYTEST_NAME;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.apache.phoenix.util.TestUtil.createGroupByTestTable;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -33,6 +33,7 @@ import java.sql.Statement;
import java.util.Map;
import java.util.Properties;
+import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.QueryUtil;
@@ -54,8 +55,10 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
// covers: COUNT, COUNT(DISTINCT) SUM, AVG, MIN, MAX
private static String GROUPBY1 = "select "
- + "count(*), count(distinct uri), sum(appcpu), avg(appcpu), uri, min(id), max(id) from "
- + GROUPBYTEST_NAME + " group by uri";
+ + "count(*), count(distinct uri), sum(appcpu), avg(appcpu), uri, min(id), max(id) from %s "
+ + "group by uri";
+
+ private static String GROUPBY2 = "select count(distinct uri) from %s";
private int id;
@@ -84,10 +87,21 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
createGroupByTestTable(conn, tableName);
}
- private void loadData(Connection conn) throws SQLException {
+ private void loadData(Connection conn, String tableName) throws SQLException {
int groupFactor = NUM_ROWS_INSERTED / 2;
for (int i = 0; i < NUM_ROWS_INSERTED; i++) {
- insertRow(conn, Integer.toString(i % (groupFactor)), 10);
+ insertRow(conn, tableName, Integer.toString(i % (groupFactor)), 10);
+
+ if ((i % 1000) == 0) {
+ conn.commit();
+ }
+ }
+ conn.commit();
+ }
+
+ private void loadUniqueURIData(Connection conn, String tableName, int rowsToInsert) throws SQLException {
+ for (int i = 0; i < rowsToInsert; i++) {
+ insertRow(conn, tableName, Integer.toString(i), 10);
if ((i % 1000) == 0) {
conn.commit();
@@ -96,10 +110,10 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
conn.commit();
}
- private void insertRow(Connection conn, String uri, int appcpu)
+ private void insertRow(Connection conn, String tableName, String uri, int appcpu)
throws SQLException {
PreparedStatement statement = conn.prepareStatement("UPSERT INTO "
- + GROUPBYTEST_NAME + "(id, uri, appcpu) values (?,?,?)");
+ + tableName + "(id, uri, appcpu) values (?,?,?)");
statement.setString(1, String.valueOf(id));
statement.setString(2, uri);
statement.setInt(3, appcpu);
@@ -110,14 +124,14 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
@Test
public void testScanUri() throws Exception {
- SpillableGroupByIT spGpByT = new SpillableGroupByIT();
+ String tableName = generateUniqueName();
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
Connection conn = DriverManager.getConnection(getUrl(), props);
- createTable(conn, GROUPBYTEST_NAME);
- spGpByT.loadData(conn);
+ createTable(conn, tableName);
+ loadData(conn, tableName);
props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
Statement stmt = conn.createStatement();
- ResultSet rs = stmt.executeQuery(GROUPBY1);
+ ResultSet rs = stmt.executeQuery(String.format(GROUPBY1, tableName));
int count = 0;
while (rs.next()) {
@@ -135,14 +149,14 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
assertEquals(NUM_ROWS_INSERTED / 2, count);
conn.createStatement();
- rs = stmt.executeQuery("SELECT appcpu FROM " + GROUPBYTEST_NAME + " group by appcpu limit 1");
+ rs = stmt.executeQuery("SELECT appcpu FROM " + tableName + " group by appcpu limit 1");
assertTrue(rs.next());
assertEquals(10,rs.getInt(1));
assertFalse(rs.next());
stmt = conn.createStatement();
- rs = stmt.executeQuery("SELECT to_number(uri) FROM " + GROUPBYTEST_NAME + " group by to_number(uri) limit 100");
+ rs = stmt.executeQuery("SELECT to_number(uri) FROM " + tableName + " group by to_number(uri) limit 100");
count = 0;
while (rs.next()) {
count++;
@@ -152,24 +166,44 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
@Test
public void testStatisticsAreNotWritten() throws SQLException {
+ String tableName = generateUniqueName();
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
Connection conn = DriverManager.getConnection(getUrl(), props);
Statement stmt = conn.createStatement();
- stmt.execute("CREATE TABLE T1 (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR)");
- stmt.execute("UPSERT INTO T1 VALUES (1, 'NAME1')");
- stmt.execute("UPSERT INTO T1 VALUES (2, 'NAME2')");
- stmt.execute("UPSERT INTO T1 VALUES (3, 'NAME3')");
+ stmt.execute("CREATE TABLE " + tableName + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR)");
+ stmt.execute("UPSERT INTO " + tableName + " VALUES (1, 'NAME1')");
+ stmt.execute("UPSERT INTO " + tableName + " VALUES (2, 'NAME2')");
+ stmt.execute("UPSERT INTO " + tableName + " VALUES (3, 'NAME3')");
conn.commit();
- stmt.execute("UPDATE STATISTICS T1");
+ stmt.execute("UPDATE STATISTICS " + tableName);
ResultSet rs = stmt.executeQuery("SELECT * FROM \"SYSTEM\".STATS");
assertFalse(rs.next());
rs.close();
stmt.close();
- rs = conn.createStatement().executeQuery("EXPLAIN SELECT * FROM T1");
+ rs = conn.createStatement().executeQuery("EXPLAIN SELECT * FROM " + tableName);
String explainPlan = QueryUtil.getExplainPlan(rs);
assertEquals(
- "CLIENT 1-CHUNK PARALLEL 1-WAY FULL SCAN OVER T1",
+ "CLIENT 1-CHUNK PARALLEL 1-WAY FULL SCAN OVER " + tableName,
explainPlan);
conn.close();
}
+
+ @Test
+ public void testDistinctCountFails() throws Exception {
+ String tableName = generateUniqueName();
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ createTable(conn, tableName);
+ loadUniqueURIData(conn, tableName, 1000);
+ props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(String.format(GROUPBY2, tableName));
+ try {
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.INSUFFICIENT_MEMORY.getErrorCode(),e.getErrorCode());
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 6108aca..72ca58d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -76,7 +76,9 @@ import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.WritableUtils;
+import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.ServerCacheClient;
+import org.apache.phoenix.cache.TenantCache;
import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.exception.DataExceedsCapacityException;
import org.apache.phoenix.exception.SQLExceptionCode;
@@ -98,6 +100,7 @@ import org.apache.phoenix.index.PhoenixIndexFailurePolicy;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy.MutateCommand;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.join.HashJoinInfo;
+import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -515,7 +518,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
byte[] clientVersionBytes = scan.getAttribute(PhoenixIndexCodec.CLIENT_VERSION);
boolean acquiredLock = false;
boolean incrScanRefCount = false;
- try {
+ final TenantCache tenantCache = GlobalCache.getTenantCache(env, ScanUtil.getTenantId(scan));
+ try (MemoryChunk em = tenantCache.getMemoryManager().allocate(0)) {
if(needToWrite) {
synchronized (lock) {
if (isRegionClosingOrSplitting) {
@@ -528,6 +532,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
}
region.startRegionOperation();
acquiredLock = true;
+ long size = 0;
synchronized (innerScanner) {
do {
List<Cell> results = useQualifierAsIndex ? new EncodedColumnQualiferCellsList(minMaxQualifiers.getFirst(), minMaxQualifiers.getSecond(), encodingScheme) : new ArrayList<Cell>();
@@ -759,7 +764,11 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
commitBatch(region, indexMutations, blockingMemStoreSize);
indexMutations.clear();
}
- aggregators.aggregate(rowAggregators, result);
+ size += aggregators.aggregate(rowAggregators, result);
+ while(size > em.getSize()) {
+ logger.info("Request: {}, resizing {} by 1024*1024", size, em.getSize());
+ em.resize(em.getSize() + 1024*1024);
+ }
hasAny = true;
}
} while (hasMore);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index dcf761a..9cbc67e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -434,14 +434,15 @@ public enum SQLExceptionCode {
SCHEMA_NOT_ALLOWED(724, "43M07", "Schema name not allowed!!"),
CREATE_SCHEMA_NOT_ALLOWED(725, "43M08", "Cannot create schema because config "
+ QueryServices.IS_NAMESPACE_MAPPING_ENABLED + " for enabling name space mapping isn't enabled."),
- INCONSISTENT_NAMESPACE_MAPPING_PROPERTIES(726, "43M10", " Inconsistent namespace mapping properties.."),
- ASYNC_NOT_ALLOWED(727, "43M11", " ASYNC option is not allowed.. "),
+ INCONSISTENT_NAMESPACE_MAPPING_PROPERTIES(726, "43M10", " Inconsistent namespace mapping properties."),
+ ASYNC_NOT_ALLOWED(727, "43M11", " ASYNC option is not allowed."),
NEW_CONNECTION_THROTTLED(728, "410M1", "Could not create connection " +
"because this client already has the maximum number" +
" of connections to the target cluster."),
MAX_MUTATION_SIZE_EXCEEDED(729, "LIM01", "MutationState size is bigger than maximum allowed number of rows"),
MAX_MUTATION_SIZE_BYTES_EXCEEDED(730, "LIM02", "MutationState size is bigger than maximum allowed number of bytes"),
+ INSUFFICIENT_MEMORY(999, "50M01", "Unable to allocate enough memory."),
HASH_JOIN_CACHE_NOT_FOUND(900, "HJ01", "Hash Join cache not found");
private final int errorCode;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/Aggregators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/Aggregators.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/Aggregators.java
index b1dc658..2f6e6ee 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/Aggregators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/Aggregators.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.expression.aggregator;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.expression.function.SingleAggregateFunction;
+import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
import org.apache.phoenix.schema.KeyValueSchema;
import org.apache.phoenix.schema.KeyValueSchema.KeyValueSchemaBuilder;
import org.apache.phoenix.schema.ValueBitSet;
@@ -82,7 +83,7 @@ abstract public class Aggregators {
* Aggregate over aggregators
* @param result the single row Result from scan iteration
*/
- abstract public void aggregate(Aggregator[] aggregators, Tuple result);
+ abstract public long aggregate(Aggregator[] aggregators, Tuple result);
protected static int calculateSize(Aggregator[] aggregators) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ClientAggregators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ClientAggregators.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ClientAggregators.java
index 54d5690..f1ed2a9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ClientAggregators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ClientAggregators.java
@@ -50,7 +50,7 @@ public class ClientAggregators extends Aggregators {
}
@Override
- public void aggregate(Aggregator[] aggregators, Tuple result) {
+ public long aggregate(Aggregator[] aggregators, Tuple result) {
TupleUtil.getAggregateValue(result, ptr);
tempValueSet.clear();
tempValueSet.or(ptr);
@@ -64,6 +64,7 @@ public class ClientAggregators extends Aggregators {
}
i++;
}
+ return 0;
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/DistinctValueWithCountServerAggregator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/DistinctValueWithCountServerAggregator.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/DistinctValueWithCountServerAggregator.java
index 4801a9d..ea7474f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/DistinctValueWithCountServerAggregator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/DistinctValueWithCountServerAggregator.java
@@ -52,6 +52,7 @@ public class DistinctValueWithCountServerAggregator extends BaseAggregator {
private static final int FIXED_COPY_THRESHOLD = SizedUtil.ARRAY_SIZE * 2;
private int compressThreshold;
+ private int heapSize = 0;
private byte[] buffer = null;
protected Map<ImmutableBytesPtr, Integer> valueVsCount = new HashMap<ImmutableBytesPtr, Integer>();
@@ -75,6 +76,9 @@ public class DistinctValueWithCountServerAggregator extends BaseAggregator {
Integer count = this.valueVsCount.get(key);
if (count == null) {
this.valueVsCount.put(key, 1);
+ heapSize += SizedUtil.MAP_ENTRY_SIZE + // entry
+ Bytes.SIZEOF_INT + // key size
+ key.getLength() + SizedUtil.ARRAY_SIZE; // value size
} else {
this.valueVsCount.put(key, ++count);
}
@@ -130,20 +134,7 @@ public class DistinctValueWithCountServerAggregator extends BaseAggregator {
// The heap size which will be taken by the count map.
private int countMapHeapSize() {
- int size = 0;
- if (this.valueVsCount.size() > 0) {
- for (ImmutableBytesPtr key : this.valueVsCount.keySet()) {
- size += SizedUtil.MAP_ENTRY_SIZE + // entry
- Bytes.SIZEOF_INT + // key size
- key.getLength() + SizedUtil.ARRAY_SIZE; // value size
- }
- } else {
- // Initially when the getSize() is called, we dont have any entries in the map so as to
- // tell the exact heap need. Let us approximate the #entries
- SizedUtil.sizeOfMap(DEFAULT_ESTIMATED_DISTINCT_VALUES,
- SizedUtil.IMMUTABLE_BYTES_PTR_SIZE, Bytes.SIZEOF_INT);
- }
- return size;
+ return heapSize;
}
@Override
@@ -154,6 +145,7 @@ public class DistinctValueWithCountServerAggregator extends BaseAggregator {
@Override
public void reset() {
valueVsCount = new HashMap<ImmutableBytesPtr, Integer>();
+ heapSize = 0;
buffer = null;
super.reset();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ServerAggregators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ServerAggregators.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ServerAggregators.java
index 366bbc6..790939c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ServerAggregators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/ServerAggregators.java
@@ -52,13 +52,17 @@ public class ServerAggregators extends Aggregators {
}
@Override
- public void aggregate(Aggregator[] aggregators, Tuple result) {
+ public long aggregate(Aggregator[] aggregators, Tuple result) {
+ long dsize = 0;
for (int i = 0; i < expressions.length; i++) {
if (expressions[i].evaluate(result, ptr) && ptr.getLength() != 0) {
+ dsize -= aggregators[i].getSize();
aggregators[i].aggregate(result, ptr);
+ dsize += aggregators[i].getSize();
}
expressions[i].reset();
}
+ return dsize;
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/memory/ChildMemoryManager.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/memory/ChildMemoryManager.java b/phoenix-core/src/main/java/org/apache/phoenix/memory/ChildMemoryManager.java
index 8f7571b..da009fb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/memory/ChildMemoryManager.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/memory/ChildMemoryManager.java
@@ -19,6 +19,8 @@ package org.apache.phoenix.memory;
import org.apache.http.annotation.GuardedBy;
import org.apache.http.annotation.ThreadSafe;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
/**
*
@@ -54,7 +56,10 @@ public class ChildMemoryManager extends DelegatingMemoryManager {
long availBytes = getAvailableMemory();
// Check if this memory managers percentage of allocated bytes exceeds its allowed maximum
if (minBytes > availBytes) {
- throw new InsufficientMemoryException("Attempt to allocate more memory than the max allowed of " + maxPercOfTotal + "%");
+ throw new InsufficientMemoryException(
+ new SQLExceptionInfo.Builder(SQLExceptionCode.INSUFFICIENT_MEMORY)
+ .setMessage("Attempt to allocate more memory than the max allowed of " + maxPercOfTotal + "%")
+ .build().buildException());
}
// Revise reqBytes down to available memory if necessary
return Math.min(reqBytes,availBytes);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/main/java/org/apache/phoenix/memory/GlobalMemoryManager.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/memory/GlobalMemoryManager.java b/phoenix-core/src/main/java/org/apache/phoenix/memory/GlobalMemoryManager.java
index 651526f..fe0d6d7 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/memory/GlobalMemoryManager.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/memory/GlobalMemoryManager.java
@@ -18,6 +18,8 @@
package org.apache.phoenix.memory;
import org.apache.http.annotation.GuardedBy;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -55,7 +57,6 @@ public class GlobalMemoryManager implements MemoryManager {
return maxMemoryBytes;
}
-
// TODO: Work on fairness: One big memory request can cause all others to fail here.
private long allocateBytes(long minBytes, long reqBytes) {
if (minBytes < 0 || reqBytes < 0) {
@@ -63,15 +64,21 @@ public class GlobalMemoryManager implements MemoryManager {
+ ") and requested bytes (" + reqBytes + ") must be greater than zero");
}
if (minBytes > maxMemoryBytes) {
- throw new InsufficientMemoryException("Requested memory of " + minBytes
- + " bytes is larger than global pool of " + maxMemoryBytes + " bytes.");
+ throw new InsufficientMemoryException(
+ new SQLExceptionInfo.Builder(SQLExceptionCode.INSUFFICIENT_MEMORY)
+ .setMessage("Requested memory of " + minBytes
+ + " bytes is larger than global pool of " + maxMemoryBytes + " bytes.")
+ .build().buildException());
}
long nBytes;
synchronized(sync) {
if (usedMemoryBytes + minBytes > maxMemoryBytes) {
- throw new InsufficientMemoryException("Requested memory of " + minBytes
- + " bytes could not be allocated. Using memory of " + usedMemoryBytes
- + " bytes from global pool of " + maxMemoryBytes);
+ throw new InsufficientMemoryException(
+ new SQLExceptionInfo.Builder(SQLExceptionCode.INSUFFICIENT_MEMORY)
+ .setMessage("Requested memory of " + minBytes
+ + " bytes could not be allocated. Using memory of " + usedMemoryBytes
+ + " bytes from global pool of " + maxMemoryBytes)
+ .build().buildException());
}
// Allocate at most reqBytes, but at least minBytes
nBytes = Math.min(reqBytes, maxMemoryBytes - usedMemoryBytes);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/45f856af/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java b/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
index d50589f..1ec07b6 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
@@ -228,7 +228,6 @@ public class TestUtil {
public static final String STABLE_NAME = "STABLE";
public static final String STABLE_PK_NAME = "ID";
public static final String STABLE_SCHEMA_NAME = "";
- public static final String GROUPBYTEST_NAME = "GROUPBYTEST";
public static final String CUSTOM_ENTITY_DATA_FULL_NAME = "CORE.CUSTOM_ENTITY_DATA";
public static final String CUSTOM_ENTITY_DATA_NAME = "CUSTOM_ENTITY_DATA";
public static final String CUSTOM_ENTITY_DATA_SCHEMA_NAME = "CORE";
[4/4] phoenix git commit: PHOENIX-4644 Array modification functions
should require two arguments
Posted by ja...@apache.org.
PHOENIX-4644 Array modification functions should require two arguments
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d2715101
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d2715101
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d2715101
Branch: refs/heads/master
Commit: d2715101e10ce2a5b3dfbaf8a8f23997fbc6ec58
Parents: abfc1ff
Author: James Taylor <jt...@salesforce.com>
Authored: Fri Mar 9 15:00:03 2018 -0800
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Mar 9 15:20:04 2018 -0800
----------------------------------------------------------------------
.../phoenix/end2end/ArrayAppendFunctionIT.java | 43 ++++++++++++---
.../function/ArrayAppendFunction.java | 11 +++-
.../function/ArrayConcatFunction.java | 3 +-
.../function/ArrayModifierFunction.java | 55 ++++++++++++++++---
.../function/ArrayPrependFunction.java | 9 ++-
.../phoenix/parse/ArrayModifierParseNode.java | 36 ++++++++++++
.../apache/phoenix/parse/FunctionParseNode.java | 12 +++-
.../org/apache/phoenix/util/ExpressionUtil.java | 2 +-
.../phoenix/compile/QueryCompilerTest.java | 58 +++++++++++++++++++-
9 files changed, 202 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/it/java/org/apache/phoenix/end2end/ArrayAppendFunctionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ArrayAppendFunctionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ArrayAppendFunctionIT.java
index 089cd41..caa17fe 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ArrayAppendFunctionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ArrayAppendFunctionIT.java
@@ -81,26 +81,53 @@ public class ArrayAppendFunctionIT extends ParallelStatsDisabledIT {
String tableName = initTables(conn);
ResultSet rs;
- rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(nullVarChar,'34567'),ARRAY_PREPEND('34567',nullVarChar) FROM " + tableName + " LIMIT 1");
- assertTrue(rs.next());
-
String[] strings = new String[]{"34567"};
-
Array array = conn.createArrayOf("VARCHAR", strings);
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(null,'34567') FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
+
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_PREPEND('34567',null) FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
+
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(nullVarChar,'34567') FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
assertEquals(array, rs.getArray(1));
- assertEquals(array, rs.getArray(2));
assertFalse(rs.next());
- rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(nullBigint,123),ARRAY_PREPEND(123,nullBigint) FROM " + tableName + " LIMIT 1");
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_PREPEND('34567',nullVarChar) FROM " + tableName + " LIMIT 1");
assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
- Long[] longs = new Long[]{123L};
+ Integer[] ints = new Integer[]{123};
+ array = conn.createArrayOf("INTEGER", ints);
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(null,123) FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
+
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_PREPEND(123,null) FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
+
+ Long[] longs = new Long[]{123L};
array = conn.createArrayOf("BIGINT", longs);
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_APPEND(nullBigint,123) FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
+ assertEquals(array, rs.getArray(1));
+ assertFalse(rs.next());
+
+ rs = conn.createStatement().executeQuery("SELECT ARRAY_PREPEND(123,nullBigint) FROM " + tableName + " LIMIT 1");
+ assertTrue(rs.next());
assertEquals(array, rs.getArray(1));
- assertEquals(array, rs.getArray(2));
assertFalse(rs.next());
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayAppendFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayAppendFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayAppendFunction.java
index 8c7fa9f..d0c0b3b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayAppendFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayAppendFunction.java
@@ -21,13 +21,18 @@ import java.util.List;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.ArrayModifierParseNode;
import org.apache.phoenix.parse.FunctionParseNode;
import org.apache.phoenix.schema.TypeMismatchException;
-import org.apache.phoenix.schema.types.*;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PBinaryArray;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.schema.types.PVarbinaryArray;
-@FunctionParseNode.BuiltInFunction(name = ArrayAppendFunction.NAME, args = {
+@FunctionParseNode.BuiltInFunction(name = ArrayAppendFunction.NAME, nodeClass=ArrayModifierParseNode.class, args = {
@FunctionParseNode.Argument(allowedTypes = {PBinaryArray.class, PVarbinaryArray.class}),
- @FunctionParseNode.Argument(allowedTypes = {PVarbinary.class}, defaultValue = "null")})
+ @FunctionParseNode.Argument(allowedTypes = {PVarbinary.class})})
public class ArrayAppendFunction extends ArrayModifierFunction {
public static final String NAME = "ARRAY_APPEND";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayConcatFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayConcatFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayConcatFunction.java
index 28774cb..5e4dba9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayConcatFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayConcatFunction.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.ArrayModifierParseNode;
import org.apache.phoenix.parse.FunctionParseNode;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TypeMismatchException;
@@ -30,7 +31,7 @@ import org.apache.phoenix.schema.types.PBinaryArray;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PVarbinaryArray;
-@FunctionParseNode.BuiltInFunction(name = ArrayConcatFunction.NAME, args = {
+@FunctionParseNode.BuiltInFunction(name = ArrayConcatFunction.NAME, nodeClass=ArrayModifierParseNode.class, args = {
@FunctionParseNode.Argument(allowedTypes = {PBinaryArray.class, PVarbinaryArray.class}),
@FunctionParseNode.Argument(allowedTypes = {PBinaryArray.class, PVarbinaryArray.class})})
public class ArrayConcatFunction extends ArrayModifierFunction {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayModifierFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayModifierFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayModifierFunction.java
index 2039690..b53ca85 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayModifierFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayModifierFunction.java
@@ -18,6 +18,8 @@
package org.apache.phoenix.expression.function;
+import java.io.DataInput;
+import java.io.IOException;
import java.util.List;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -29,8 +31,10 @@ import org.apache.phoenix.schema.TypeMismatchException;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PArrayDataType;
import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.ExpressionUtil;
public abstract class ArrayModifierFunction extends ScalarFunction {
+ private boolean isNullArray;
public ArrayModifierFunction() {
}
@@ -72,6 +76,16 @@ public abstract class ArrayModifierFunction extends ScalarFunction {
throw new DataExceedsCapacityException(baseDataType, arrayExpr.getMaxLength(),
arrayExpr.getScale());
}
+ init();
+ }
+
+ private void init() {
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ if (getLHSExpr().getDataType().isArrayType()) {
+ isNullArray = ExpressionUtil.isNull(getLHSExpr(), ptr);
+ } else {
+ isNullArray = ExpressionUtil.isNull(getRHSExpr(), ptr);
+ }
}
@Override
@@ -142,7 +156,8 @@ public abstract class ArrayModifierFunction extends ScalarFunction {
public PDataType getLHSBaseType() {
if (getLHSExpr().getDataType().isArrayType()) {
- return PDataType.arrayBaseType(getLHSExpr().getDataType());
+ // Use RHS type if we have a null constant to get the correct array type
+ return isNullArray ? getRHSExpr().getDataType() : PDataType.arrayBaseType(getLHSExpr().getDataType());
} else {
return getLHSExpr().getDataType();
}
@@ -150,7 +165,8 @@ public abstract class ArrayModifierFunction extends ScalarFunction {
public PDataType getRHSBaseType() {
if (getRHSExpr().getDataType().isArrayType()) {
- return PDataType.arrayBaseType(getRHSExpr().getDataType());
+ // Use LHS type if we have a null constant to get the correct array type
+ return isNullArray ? getLHSExpr().getDataType() : PDataType.arrayBaseType(getRHSExpr().getDataType());
} else {
return getRHSExpr().getDataType();
}
@@ -159,29 +175,52 @@ public abstract class ArrayModifierFunction extends ScalarFunction {
@Override
public PDataType getDataType() {
if (getLHSExpr().getDataType().isArrayType()) {
- return getLHSExpr().getDataType();
+ // Use array of RHS type if we have a null constant since otherwise we'd use binary
+ return isNullArray ? getRHSExpr().getDataType().isArrayType() ? getRHSExpr().getDataType() : PDataType.fromTypeId(getRHSExpr().getDataType().getSqlType() + PDataType.ARRAY_TYPE_BASE) : getLHSExpr().getDataType();
} else {
- return getRHSExpr().getDataType();
+ return isNullArray ? getLHSExpr().getDataType().isArrayType() ? getLHSExpr().getDataType() : PDataType.fromTypeId(getLHSExpr().getDataType().getSqlType() + PDataType.ARRAY_TYPE_BASE) : getRHSExpr().getDataType();
}
}
+ private Integer getMaxLength(Expression expression) {
+ PDataType type = expression.getDataType();
+ if (type.isFixedWidth() && type.getByteSize() != null) {
+ return type.getByteSize();
+ }
+ return expression.getMaxLength();
+ }
@Override
public Integer getMaxLength() {
if (getLHSExpr().getDataType().isArrayType()) {
- return getLHSExpr().getMaxLength();
+ // Use max length of RHS if we have a null constant since otherwise we'd use null (which breaks fixed types)
+ return getMaxLength(isNullArray ? getRHSExpr() : getLHSExpr());
} else {
- return getRHSExpr().getMaxLength();
+ return getMaxLength(isNullArray ? getLHSExpr() : getRHSExpr());
}
}
@Override
public SortOrder getSortOrder() {
if (getLHSExpr().getDataType().isArrayType()) {
- return getLHSExpr().getSortOrder();
+ return isNullArray ? getRHSExpr().getSortOrder() : getLHSExpr().getSortOrder();
} else {
- return getRHSExpr().getSortOrder();
+ return isNullArray ? getLHSExpr().getSortOrder() : getRHSExpr().getSortOrder();
}
}
+ @Override
+ public Integer getScale() {
+ if (getLHSExpr().getDataType().isArrayType()) {
+ return isNullArray ? getRHSExpr().getScale() : getLHSExpr().getScale();
+ } else {
+ return isNullArray ? getLHSExpr().getScale() : getRHSExpr().getScale();
+ }
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ super.readFields(input);
+ init();
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayPrependFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayPrependFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayPrependFunction.java
index c2311fb..41fdf2a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayPrependFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/ArrayPrependFunction.java
@@ -22,11 +22,16 @@ import java.util.List;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.ArrayModifierParseNode;
import org.apache.phoenix.parse.FunctionParseNode;
import org.apache.phoenix.schema.TypeMismatchException;
-import org.apache.phoenix.schema.types.*;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PBinaryArray;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.schema.types.PVarbinaryArray;
-@FunctionParseNode.BuiltInFunction(name = ArrayPrependFunction.NAME, args = {
+@FunctionParseNode.BuiltInFunction(name = ArrayPrependFunction.NAME, nodeClass=ArrayModifierParseNode.class, args = {
@FunctionParseNode.Argument(allowedTypes = {PVarbinary.class}),
@FunctionParseNode.Argument(allowedTypes = {PBinaryArray.class, PVarbinaryArray.class})})
public class ArrayPrependFunction extends ArrayModifierFunction {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/parse/ArrayModifierParseNode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ArrayModifierParseNode.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ArrayModifierParseNode.java
new file mode 100644
index 0000000..2ec1e27
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ArrayModifierParseNode.java
@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+
+public class ArrayModifierParseNode extends FunctionParseNode {
+
+ public ArrayModifierParseNode(String name, List<ParseNode> children, BuiltInFunctionInfo info) {
+ super(name, children, info);
+ }
+
+ @Override
+ public boolean evalToNullIfParamIsNull(StatementContext context, int index) throws SQLException {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/parse/FunctionParseNode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/FunctionParseNode.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/FunctionParseNode.java
index df9ad86..fc16156 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/FunctionParseNode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/FunctionParseNode.java
@@ -184,6 +184,10 @@ public class FunctionParseNode extends CompoundParseNode {
} else { // Use expression as is, since we already have the data type set
context.getBindManager().addParamMetaData(bindNode, child);
}
+ } else if (allowedTypes.length > 0) {
+ // Switch null type with typed null
+ children.set(i, LiteralExpression.newConstant(null, PDataTypeFactory.getInstance().instanceFromClass(
+ allowedTypes[0]), Determinism.ALWAYS));
}
} else {
validateFunctionArguement(info, i, child);
@@ -254,10 +258,14 @@ public class FunctionParseNode extends CompoundParseNode {
*/
public Expression create(List<Expression> children, PFunction function, StatementContext context) throws SQLException {
try {
+ Constructor<? extends FunctionExpression> fCtor = info.getFuncCtor();
+ if (fCtor == null) {
+ fCtor = getExpressionCtor(info.func, null);
+ }
if(function == null) {
- return info.getFuncCtor().newInstance(children);
+ return fCtor.newInstance(children);
} else {
- return info.getFuncCtor().newInstance(children, function);
+ return fCtor.newInstance(children, function);
}
} catch (InstantiationException e) {
throw new SQLException(e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/main/java/org/apache/phoenix/util/ExpressionUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ExpressionUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ExpressionUtil.java
index fbd10fc..881b0e1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ExpressionUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ExpressionUtil.java
@@ -36,7 +36,7 @@ public class ExpressionUtil {
Object value = null;
PDataType type = expression.getDataType();
if (expression.evaluate(null, ptr) && ptr.getLength() != 0) {
- value = type.toObject(ptr);
+ value = type.toObject(ptr.get(), ptr.getOffset(), ptr.getLength(), type, expression.getSortOrder(), expression.getMaxLength(), expression.getScale());
}
return LiteralExpression.newConstant(value, type, expression.getDeterminism());
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d2715101/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
index 2b5a8cb..11f5f22 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/compile/QueryCompilerTest.java
@@ -43,7 +43,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Properties;
-import com.google.common.collect.Lists;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
@@ -52,7 +51,18 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
import org.apache.phoenix.exception.SQLExceptionCode;
-import org.apache.phoenix.execute.*;
+import org.apache.phoenix.execute.AggregatePlan;
+import org.apache.phoenix.execute.ClientAggregatePlan;
+import org.apache.phoenix.execute.ClientScanPlan;
+import org.apache.phoenix.execute.CorrelatePlan;
+import org.apache.phoenix.execute.CursorFetchPlan;
+import org.apache.phoenix.execute.HashJoinPlan;
+import org.apache.phoenix.execute.LiteralResultIterationPlan;
+import org.apache.phoenix.execute.ScanPlan;
+import org.apache.phoenix.execute.SortMergeJoinPlan;
+import org.apache.phoenix.execute.TupleProjectionPlan;
+import org.apache.phoenix.execute.UnionPlan;
+import org.apache.phoenix.execute.UnnestArrayPlan;
import org.apache.phoenix.execute.visitor.QueryPlanVisitor;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.LiteralExpression;
@@ -88,6 +98,8 @@ import org.apache.phoenix.util.SchemaUtil;
import org.junit.Ignore;
import org.junit.Test;
+import com.google.common.collect.Lists;
+
/**
@@ -2177,6 +2189,48 @@ public class QueryCompilerTest extends BaseConnectionlessQueryTest {
}
@Test
+ public void testArrayAppendSingleArg() throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ try {
+ conn.createStatement().execute("CREATE TABLE t (p INTEGER PRIMARY KEY, arr1 INTEGER ARRAY, arr2 INTEGER ARRAY)");
+ conn.createStatement().executeQuery("SELECT ARRAY_APPEND(arr2) from t");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.FUNCTION_UNDEFINED.getErrorCode(),e.getErrorCode());
+ } finally {
+ conn.close();
+ }
+ }
+
+ @Test
+ public void testArrayPrependSingleArg() throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ try {
+ conn.createStatement().execute("CREATE TABLE t (p INTEGER PRIMARY KEY, arr1 INTEGER ARRAY, arr2 INTEGER ARRAY)");
+ conn.createStatement().executeQuery("SELECT ARRAY_PREPEND(arr2) from t");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.FUNCTION_UNDEFINED.getErrorCode(),e.getErrorCode());
+ } finally {
+ conn.close();
+ }
+ }
+
+ @Test
+ public void testArrayConcatSingleArg() throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ try {
+ conn.createStatement().execute("CREATE TABLE t (p INTEGER PRIMARY KEY, arr1 INTEGER ARRAY, arr2 INTEGER ARRAY)");
+ conn.createStatement().executeQuery("SELECT ARRAY_CAT(arr2) from t");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.FUNCTION_UNDEFINED.getErrorCode(),e.getErrorCode());
+ } finally {
+ conn.close();
+ }
+ }
+
+ @Test
public void testServerArrayElementProjection5() throws SQLException {
Connection conn = DriverManager.getConnection(getUrl());
try {