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 354716eed9 PHOENIX-7213 Add option for unlimited phoenix.query.QueueSize 354716eed9 is described below commit 354716eed9ef6a172ad98882194db6928385f414 Author: tkhurana <khurana.ta...@gmail.com> AuthorDate: Wed Feb 14 10:07:05 2024 -0800 PHOENIX-7213 Add option for unlimited phoenix.query.QueueSize --- .../phoenix/job/AbstractRoundRobinQueue.java | 16 +++++- .../apache/phoenix/query/QueryServicesOptions.java | 31 +++++----- .../apache/phoenix/end2end/QueryWithLimitIT.java | 67 ++++++++++++++-------- 3 files changed, 72 insertions(+), 42 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/job/AbstractRoundRobinQueue.java b/phoenix-core-client/src/main/java/org/apache/phoenix/job/AbstractRoundRobinQueue.java index fa68852524..001453923e 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/job/AbstractRoundRobinQueue.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/job/AbstractRoundRobinQueue.java @@ -17,12 +17,22 @@ */ package org.apache.phoenix.job; -import org.apache.phoenix.util.EnvironmentEdgeManager; -import java.util.*; +import java.util.AbstractQueue; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.ListIterator; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; +import org.apache.phoenix.util.EnvironmentEdgeManager; + +import static org.apache.phoenix.query.QueryServicesOptions.UNLIMITED_QUEUE_SIZE; + /** * * An bounded blocking queue implementation that keeps a virtual queue of elements on per-producer @@ -88,7 +98,7 @@ public abstract class AbstractRoundRobinQueue<E> extends AbstractQueue<E> ProducerList<E> producerList = null; synchronized(lock) { - if (this.size == this.maxSize) { + if (this.maxSize != UNLIMITED_QUEUE_SIZE && this.size == this.maxSize) { return false; } producerList = this.producerMap.get(producerKey); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java index f88ec01a5a..2d8d796a4f 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java @@ -141,20 +141,21 @@ import org.apache.phoenix.util.ReadOnlyProps; * @since 0.1 */ public class QueryServicesOptions { - public static final int DEFAULT_KEEP_ALIVE_MS = 60000; - public static final int DEFAULT_THREAD_POOL_SIZE = 128; - public static final int DEFAULT_QUEUE_SIZE = 5000; - public static final int DEFAULT_THREAD_TIMEOUT_MS = 600000; // 10min - public static final int DEFAULT_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m - public static final int DEFAULT_SERVER_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m - public static final int DEFAULT_CLIENT_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m - public static final boolean DEFAULT_CLIENT_ORDERBY_SPOOLING_ENABLED = true; - public static final boolean DEFAULT_CLIENT_JOIN_SPOOLING_ENABLED = true; - public static final boolean DEFAULT_SERVER_ORDERBY_SPOOLING_ENABLED = true; + public static final int DEFAULT_KEEP_ALIVE_MS = 60000; + public static final int DEFAULT_THREAD_POOL_SIZE = 128; + public static final int DEFAULT_QUEUE_SIZE = 5000; + public static final int UNLIMITED_QUEUE_SIZE = -1; + public static final int DEFAULT_THREAD_TIMEOUT_MS = 600000; // 10min + public static final int DEFAULT_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m + public static final int DEFAULT_SERVER_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m + public static final int DEFAULT_CLIENT_SPOOL_THRESHOLD_BYTES = 1024 * 1024 * 20; // 20m + public static final boolean DEFAULT_CLIENT_ORDERBY_SPOOLING_ENABLED = true; + public static final boolean DEFAULT_CLIENT_JOIN_SPOOLING_ENABLED = true; + public static final boolean DEFAULT_SERVER_ORDERBY_SPOOLING_ENABLED = true; public static final String DEFAULT_SPOOL_DIRECTORY = System.getProperty("java.io.tmpdir"); - public static final int DEFAULT_MAX_MEMORY_PERC = 15; // 15% of heap - public static final int DEFAULT_MAX_TENANT_MEMORY_PERC = 100; - public static final long DEFAULT_MAX_SERVER_CACHE_SIZE = 1024*1024*100; // 100 Mb + public static final int DEFAULT_MAX_MEMORY_PERC = 15; // 15% of heap + public static final int DEFAULT_MAX_TENANT_MEMORY_PERC = 100; + public static final long DEFAULT_MAX_SERVER_CACHE_SIZE = 1024 * 1024 * 100; // 100 Mb public static final int DEFAULT_TARGET_QUERY_CONCURRENCY = 32; public static final int DEFAULT_MAX_QUERY_CONCURRENCY = 64; public static final String DEFAULT_DATE_FORMAT = DateUtil.DEFAULT_DATE_FORMAT; @@ -178,7 +179,7 @@ public class QueryServicesOptions { public final static int DEFAULT_MUTATE_BATCH_SIZE = 100; // Batch size for UPSERT SELECT and DELETE //Batch size in bytes for UPSERT, SELECT and DELETE. By default, 2MB public final static long DEFAULT_MUTATE_BATCH_SIZE_BYTES = 2097152; - // The only downside of it being out-of-sync is that the parallelization of the scan won't be as balanced as it could be. + // The only downside of it being out-of-sync is that the parallelization of the scan won't be as balanced as it could be. public static final int DEFAULT_MAX_SERVER_CACHE_TIME_TO_LIVE_MS = 30000; // 30 sec (with no activity) public static final int DEFAULT_MAX_SERVER_CACHE_PERSISTENCE_TIME_TO_LIVE_MS = 30 * 60000; // 30 minutes public static final int DEFAULT_SCAN_CACHE_SIZE = 1000; @@ -406,7 +407,7 @@ public class QueryServicesOptions { public static final int DEFAULT_LOG_SALT_BUCKETS = 32; public static final int DEFAULT_SALT_BUCKETS = 0; - public static final boolean DEFAULT_SYSTEM_CATALOG_SPLITTABLE = true; + public static final boolean DEFAULT_SYSTEM_CATALOG_SPLITTABLE = true; public static final String DEFAULT_GUIDE_POSTS_CACHE_FACTORY_CLASS = "org.apache.phoenix.query.DefaultGuidePostsCacheFactory"; diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java index 2f778c0fc6..377b6d7cfc 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java @@ -17,6 +17,8 @@ */ package org.apache.phoenix.end2end; +import static org.apache.phoenix.query.QueryServicesOptions.UNLIMITED_QUEUE_SIZE; +import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -36,6 +38,7 @@ import org.apache.phoenix.compile.ExplainPlanAttributes; import org.apache.phoenix.jdbc.PhoenixPreparedStatement; import org.apache.phoenix.query.BaseTest; import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.ReadOnlyProps; import org.apache.phoenix.util.TestUtil; @@ -49,25 +52,25 @@ import org.apache.phoenix.thirdparty.com.google.common.collect.Maps; public class QueryWithLimitIT extends BaseTest { private String tableName; - - @Before - public void generateTableName() { - tableName = generateUniqueName(); - } - + private static Map<String,String> props = Maps.newHashMapWithExpectedSize(5); @BeforeClass public static synchronized void doSetup() throws Exception { - Map<String,String> props = Maps.newHashMapWithExpectedSize(3); // Must update config before starting server props.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(50)); props.put(QueryServices.QUEUE_SIZE_ATTRIB, Integer.toString(1)); props.put(QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB, Integer.toString(0)); // Prevents RejectedExecutionException when creatomg sequence table props.put(QueryServices.THREAD_POOL_SIZE_ATTRIB, Integer.toString(4)); props.put(QueryServices.LOG_SALT_BUCKETS_ATTRIB, Integer.toString(0)); // Prevents RejectedExecutionException when creating log table + } + + @Before + public void setupDriver() throws Exception { + destroyDriver(); setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator())); + tableName = generateUniqueName(); } - + @Test public void testQueryWithLimitAndStats() throws Exception { Properties props = PropertiesUtil.deepCopy(TestUtil.TEST_PROPERTIES); @@ -106,26 +109,42 @@ public class QueryWithLimitIT extends BaseTest { @Test public void testQueryWithoutLimitFails() throws Exception { - Properties props = PropertiesUtil.deepCopy(TestUtil.TEST_PROPERTIES); - Connection conn = DriverManager.getConnection(getUrl(), props); - - conn.createStatement().execute("create table " + tableName + "\n" + - " (i1 integer not null, i2 integer not null\n" + - " CONSTRAINT pk PRIMARY KEY (i1,i2))"); - initTableValues(conn, 100); - conn.createStatement().execute("UPDATE STATISTICS " + tableName); - + Properties connProps = PropertiesUtil.deepCopy(TestUtil.TEST_PROPERTIES); String query = "SELECT i1 FROM " + tableName; + try (Connection conn = DriverManager.getConnection(getUrl(), connProps)) { + + conn.createStatement().execute("create table " + tableName + "\n" + + " (i1 integer not null, i2 integer not null\n" + + " CONSTRAINT pk PRIMARY KEY (i1,i2))"); + initTableValues(conn, 100); + conn.createStatement().execute("UPDATE STATISTICS " + tableName); + + try { + ResultSet rs = conn.createStatement().executeQuery(query); + rs.next(); + fail(); + } catch (SQLException e) { + assertTrue(e.getCause() instanceof RejectedExecutionException); + } + } + + // now run the same test with queue size set to unlimited try { - ResultSet rs = conn.createStatement().executeQuery(query); - rs.next(); - fail(); - } catch (SQLException e) { - assertTrue(e.getCause() instanceof RejectedExecutionException); + destroyDriver(); + // copy the existing properties + Map<String, String> newProps = Maps.newHashMap(props); + newProps.put(QueryServices.QUEUE_SIZE_ATTRIB, Integer.toString(UNLIMITED_QUEUE_SIZE)); + setUpTestDriver(new ReadOnlyProps(newProps.entrySet().iterator())); + try (Connection conn = DriverManager.getConnection(getUrl(), connProps)) { + // now the query should succeed + ResultSet rs = conn.createStatement().executeQuery(query); + assertTrue(rs.next()); + } + } finally { + destroyDriver(); } - conn.close(); } - + protected void initTableValues(Connection conn, int nRows) throws Exception { PreparedStatement stmt = conn.prepareStatement( "upsert into " + tableName +