This is an automated email from the ASF dual-hosted git repository.
kadir pushed a commit to branch 5.1
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/5.1 by this push:
new 1c99e01309 Revert "PHOENIX-6776 :- Abort scans of closed connections
at ScanningResultIterator (#1517) (#1525)"
1c99e01309 is described below
commit 1c99e013091a57f29636f90e40f65a9d9ffeaa93
Author: Kadir Ozdemir <[email protected]>
AuthorDate: Mon Nov 21 11:52:43 2022 -0800
Revert "PHOENIX-6776 :- Abort scans of closed connections at
ScanningResultIterator (#1517) (#1525)"
This reverts commit caa436659b2bc8e520d6c9492bc768dcecc544ac.
---
.../end2end/PreMatureTimelyAbortScanIt.java | 89 ----------------------
.../coprocessor/BaseScannerRegionObserver.java | 2 +-
.../apache/phoenix/exception/SQLExceptionCode.java | 5 +-
.../phoenix/iterate/BaseResultIterators.java | 3 -
.../phoenix/iterate/ScanningResultIterator.java | 26 +------
.../phoenix/iterate/TableResultIterator.java | 2 +-
.../iterate/TableSnapshotResultIterator.java | 7 +-
.../org/apache/phoenix/jdbc/PhoenixConnection.java | 23 +-----
.../phoenix/mapreduce/PhoenixRecordReader.java | 2 +-
9 files changed, 11 insertions(+), 148 deletions(-)
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PreMatureTimelyAbortScanIt.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/PreMatureTimelyAbortScanIt.java
deleted file mode 100644
index c0729707be..0000000000
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/PreMatureTimelyAbortScanIt.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.end2end;
-
-import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
-import org.apache.phoenix.exception.SQLExceptionCode;
-import org.apache.phoenix.iterate.ScanningResultIterator;
-import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.query.QueryServices;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
-import org.apache.phoenix.util.ReadOnlyProps;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.Map;
-
-import static org.junit.Assert.*;
-@Category(ParallelStatsDisabledTest.class)
-public class PreMatureTimelyAbortScanIt extends ParallelStatsDisabledIT {
- private static final Logger LOG =
LoggerFactory.getLogger(PreMatureTimelyAbortScanIt.class);
-
- @BeforeClass
- public static synchronized void doSetup() throws Exception {
- Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
- props.put(BaseScannerRegionObserver.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
Integer.toString(60*60)); // An hour
- props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION,
Boolean.toString(false));
- props.put(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS,
Integer.toString(0));
- setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
- }
-
- private String getUniqueUrl() {
- return url + generateUniqueName();
- }
-
- @Test
- public void testPreMatureScannerAbortForCount() throws Exception {
-
- try (Connection conn = DriverManager.getConnection(getUniqueUrl())) {
- conn.createStatement().execute("CREATE TABLE LONG_BUG (ID INTEGER
PRIMARY KEY, AMOUNT DECIMAL) SALT_BUCKETS = 16 ");
- }
- try (Connection conn = DriverManager.getConnection(getUniqueUrl())) {
- for (int i = 0; i<100 ; i++) {
- int amount = -50000 + i;
- String s = "UPSERT INTO LONG_BUG (ID, AMOUNT) VALUES( " + i +
", " + amount + ")";
- conn.createStatement().execute(s);
- }
- conn.commit();
- }
-
- try {
- PhoenixConnection conn =
DriverManager.getConnection(getUniqueUrl()).unwrap(PhoenixConnection.class);
- ScanningResultIterator.setIsScannerClosedForcefully(true);
- ResultSet resultSet = conn.createStatement().executeQuery(
- "SELECT COUNT(*) FROM LONG_BUG WHERE ID % 2 = 0");
- conn.setIsClosing(true);
- resultSet.next();
- LOG.info("Count of modulus 2 for LONG_BUG :- " +
resultSet.getInt(1));
- fail("ResultSet should have been closed");
- } catch (SQLException sqe) {
-
assertEquals(SQLExceptionCode.FAILED_KNOWINGLY_FOR_TEST.getErrorCode(),
sqe.getErrorCode());
- } catch (Exception e) {
- fail();
- } finally {
- ScanningResultIterator.setIsScannerClosedForcefully(false);
- }
- }
-}
\ No newline at end of file
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index 6f8dfd49b4..2e5fbc105a 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -244,7 +244,7 @@ abstract public class BaseScannerRegionObserver extends
CompatBaseScannerRegionO
// last possible moment. You need to swap the start/stop and make
the
// start exclusive and the stop inclusive.
ScanUtil.setupReverseScan(scan);
- if (!(scan.getFilter() instanceof PagedFilter)) {
+ if (scan.getFilter() != null && !(scan.getFilter() instanceof
PagedFilter)) {
byte[] pageSizeMsBytes =
scan.getAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS);
if (pageSizeMsBytes != null) {
scan.setFilter(new PagedFilter(scan.getFilter(),
getPageSizeMsForFilter(scan)));
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 1902068d72..0a9c961efc 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
@@ -560,10 +560,7 @@ public enum SQLExceptionCode {
"CASCADE INDEX feature is not supported for local index"),
INVALID_REGION_SPLIT_POLICY(908, "43M19",
- "REGION SPLIT POLICY is incorrect."),
-
- //SQLCode for testing exceptions
- FAILED_KNOWINGLY_FOR_TEST(7777, "TEST", "Exception was thrown to test
something");
+ "REGION SPLIT POLICY is incorrect.");
private final int errorCode;
private final String sqlState;
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 45ec1f09cd..ec6e37a69e 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -51,7 +51,6 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -1415,8 +1414,6 @@ public abstract class BaseResultIterators extends
ExplainTable implements Result
}
}
- } catch (CancellationException ce) {
- LOGGER.warn("Iterator scheduled to be executed in
Future was being cancelled", ce);
}
}
addIterator(iterators, concatIterators);
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
index 0c15c67236..61924c0f99 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
@@ -53,33 +53,22 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.phoenix.compile.ExplainPlanAttributes
.ExplainPlanAttributesBuilder;
-import org.apache.phoenix.compile.StatementContext;
-import org.apache.phoenix.exception.SQLExceptionCode;
-import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.CombinableMetric;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.schema.tuple.ResultTuple;
import org.apache.phoenix.schema.tuple.Tuple;
-import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.phoenix.util.ServerUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class ScanningResultIterator implements ResultIterator {
-
- private static final Logger LOG =
LoggerFactory.getLogger(ScanningResultIterator.class);
private final ResultScanner scanner;
private final ScanMetricsHolder scanMetricsHolder;
boolean scanMetricsUpdated;
boolean scanMetricsEnabled;
- private StatementContext context;
- private static boolean throwExceptionIfScannerClosedForceFully = false;
- public ScanningResultIterator(ResultScanner scanner, Scan scan,
ScanMetricsHolder scanMetricsHolder, StatementContext context) {
+ public ScanningResultIterator(ResultScanner scanner, Scan scan,
ScanMetricsHolder scanMetricsHolder) {
this.scanner = scanner;
this.scanMetricsHolder = scanMetricsHolder;
- this.context = context;
scanMetricsUpdated = false;
scanMetricsEnabled = scan.isScanMetricsEnabled();
}
@@ -168,14 +157,6 @@ public class ScanningResultIterator implements
ResultIterator {
try {
Result result = scanner.next();
while (result != null && (result.isEmpty() || isDummy(result))) {
- if (context.getConnection().isClosing() ||
context.getConnection().isClosed()) {
- LOG.warn("Closing ResultScanner as Connection is already
closed or in middle of closing");
- if (throwExceptionIfScannerClosedForceFully) {
- throw new
SQLExceptionInfo.Builder(SQLExceptionCode.FAILED_KNOWINGLY_FOR_TEST).build().buildException();
- }
- close();
- return null;
- }
result = scanner.next();
}
if (result == null) {
@@ -207,9 +188,4 @@ public class ScanningResultIterator implements
ResultIterator {
public ResultScanner getScanner() {
return scanner;
}
-
- @VisibleForTesting
- public static void setIsScannerClosedForcefully(boolean throwException) {
- throwExceptionIfScannerClosedForceFully = throwException;
- }
}
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
index ee0bd4b7ab..5c8d6c77ce 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
@@ -236,7 +236,7 @@ public class TableResultIterator implements ResultIterator {
if (delegate == UNINITIALIZED_SCANNER) {
try {
this.scanIterator =
- new
ScanningResultIterator(htable.getScanner(scan), scan, scanMetricsHolder,
plan.getContext());
+ new
ScanningResultIterator(htable.getScanner(scan), scan, scanMetricsHolder);
} catch (IOException e) {
Closeables.closeQuietly(htable);
throw ServerUtil.parseServerException(e);
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
index 4d37b7f995..8e7430fb97 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
@@ -39,7 +39,6 @@ import
org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import
org.apache.phoenix.compile.ExplainPlanAttributes.ExplainPlanAttributesBuilder;
-import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.schema.tuple.Tuple;
@@ -74,14 +73,12 @@ public class TableSnapshotResultIterator implements
ResultIterator {
private FileSystem fs;
private int currentRegion;
private boolean closed = false;
- private StatementContext context;
- public TableSnapshotResultIterator(Configuration configuration, Scan scan,
ScanMetricsHolder scanMetricsHolder, StatementContext context)
+ public TableSnapshotResultIterator(Configuration configuration, Scan scan,
ScanMetricsHolder scanMetricsHolder)
throws IOException {
this.configuration = configuration;
this.currentRegion = -1;
this.scan = scan;
- this.context = context;
this.scanMetricsHolder = scanMetricsHolder;
this.scanIterator = UNINITIALIZED_SCANNER;
if (PhoenixConfigurationUtil.isMRSnapshotManagedExternally(configuration))
{
@@ -155,7 +152,7 @@ public class TableSnapshotResultIterator implements
ResultIterator {
RegionInfo hri = regions.get(this.currentRegion);
this.scanIterator =
new ScanningResultIterator(new SnapshotScanner(configuration, fs,
restoreDir, htd, hri, scan),
- scan, scanMetricsHolder, context);
+ scan, scanMetricsHolder);
} catch (Throwable e) {
throw ServerUtil.parseServerException(e);
}
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index efd925a138..9c927a5b2d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -161,7 +161,6 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
private int statementExecutionCounter;
private TraceScope traceScope = null;
private volatile boolean isClosed = false;
- private volatile boolean isClosing = false;
private Sampler<?> sampler;
private boolean readOnly = false;
private Consistency consistency = Consistency.STRONG;
@@ -689,7 +688,7 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
}
private void checkOpen() throws SQLException {
- if (isClosed || isClosing) {
+ if (isClosed) {
throw new SQLExceptionInfo.Builder(
SQLExceptionCode.CONNECTION_CLOSED).build()
.buildException();
@@ -698,11 +697,10 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
@Override
public void close() throws SQLException {
- if (isClosed || isClosing) {
+ if (isClosed) {
return;
}
try {
- isClosing = true;
clearMetrics();
try {
if (traceScope != null) {
@@ -713,9 +711,8 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
} finally {
services.removeConnection(this);
}
-
+
} finally {
- isClosing = false;
isClosed = true;
if(isInternalConnection()){
GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS.decrement();
@@ -920,10 +917,6 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
return isClosed;
}
- public boolean isClosing() throws SQLException {
- return isClosing;
- }
-
@Override
public boolean isReadOnly() throws SQLException {
return readOnly;
@@ -932,7 +925,7 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
@Override
public boolean isValid(int timeout) throws SQLException {
// TODO: run query here or ping
- return !isClosed && !isClosing;
+ return !isClosed;
}
@Override
@@ -1333,14 +1326,6 @@ public class PhoenixConnection implements Connection,
MetaDataMutated, SQLClosea
this.tableResultIteratorFactory = factory;
}
- /**
- * Added for testing purposes. Do not use this elsewhere.
- */
- @VisibleForTesting
- public void setIsClosing(boolean imitateIsClosing) {
- isClosing = imitateIsClosing;
- }
-
@Override
public void removeSchema(PSchema schema, long schemaTimeStamp) {
metaData.removeSchema(schema, schemaTimeStamp);
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
index d18bcffac1..91d3da40b4 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
@@ -138,7 +138,7 @@ public class PhoenixRecordReader<T extends DBWritable>
extends RecordReader<Null
if (snapshotName != null) {
// result iterator to read snapshots
final TableSnapshotResultIterator
tableSnapshotResultIterator = new TableSnapshotResultIterator(configuration,
scan,
- scanMetricsHolder, queryPlan.getContext());
+ scanMetricsHolder);
peekingResultIterator =
LookAheadResultIterator.wrap(tableSnapshotResultIterator);
LOGGER.info("Adding TableSnapshotResultIterator for scan:
" + scan);
} else {
