This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new badcdc0d131 Check for query termination before each transform
evaluation (#19907)
badcdc0d131 is described below
commit badcdc0d131b6e4f8a2097d74ee4b4122b6c6198
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Wed Nov 4 10:01:37 2026 +0100
Check for query termination before each transform evaluation (#19907)
---
.../operator/docvalsets/TransformBlockValSet.java | 29 ++++
.../TransformBlockValSetTerminationTest.java | 115 +++++++++++++++
.../TransformEvaluationDeadlineQueriesTest.java | 160 +++++++++++++++++++++
3 files changed, 304 insertions(+)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSet.java
index e7fd11a8bf9..5143126fd6f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSet.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.operator.docvalsets;
import java.math.BigDecimal;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.operator.blocks.ValueBlock;
@@ -27,6 +28,7 @@ import
org.apache.pinot.core.operator.transform.function.TransformFunction;
import org.apache.pinot.core.plan.DocIdSetPlanNode;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.query.QueryThreadContext;
import org.apache.pinot.spi.trace.InvocationRecording;
import org.apache.pinot.spi.trace.InvocationScope;
import org.apache.pinot.spi.trace.Tracing;
@@ -38,9 +40,14 @@ import org.roaringbitmap.RoaringBitmap;
/// block.
///
/// Caller is responsible for calling the correct method based on the data
source metadata for the block value set.
+///
+/// The values are not cached: each `get*Values*()` call evaluates the
transform function on the whole block again. To
+/// make sure that a caller that evaluates the transform many times in one
block (for example, once per row) cannot run
+/// past the query deadline, each evaluation first checks whether the query is
terminated.
public class TransformBlockValSet implements BlockValSet {
private final ValueBlock _valueBlock;
private final TransformFunction _transformFunction;
+ private final Supplier<String> _terminationScope;
private boolean _nullBitmapSet;
private RoaringBitmap _nullBitmap;
@@ -50,6 +57,7 @@ public class TransformBlockValSet implements BlockValSet {
public TransformBlockValSet(ValueBlock valueBlock, TransformFunction
transformFunction) {
_valueBlock = valueBlock;
_transformFunction = transformFunction;
+ _terminationScope = () -> "transform function: " +
transformFunction.getName();
}
@Nullable
@@ -89,6 +97,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public int[] getDictionaryIdsSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.INT, true);
return _transformFunction.transformToDictIdsSV(_valueBlock);
@@ -97,6 +106,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public int[] getIntValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.INT, true);
return _transformFunction.transformToIntValuesSV(_valueBlock);
@@ -105,6 +115,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public long[] getLongValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.LONG, true);
return _transformFunction.transformToLongValuesSV(_valueBlock);
@@ -113,6 +124,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public float[] getFloatValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.FLOAT, true);
return _transformFunction.transformToFloatValuesSV(_valueBlock);
@@ -121,6 +133,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public double[] getDoubleValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.DOUBLE, true);
return _transformFunction.transformToDoubleValuesSV(_valueBlock);
@@ -129,6 +142,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public BigDecimal[] getBigDecimalValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.BIG_DECIMAL, true);
return _transformFunction.transformToBigDecimalValuesSV(_valueBlock);
@@ -137,6 +151,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public String[] getStringValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.STRING, true);
return _transformFunction.transformToStringValuesSV(_valueBlock);
@@ -145,6 +160,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public byte[][] getBytesValuesSV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.BYTES, true);
return _transformFunction.transformToBytesValuesSV(_valueBlock);
@@ -186,6 +202,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public int[][] getDictionaryIdsMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.INT, false);
return _transformFunction.transformToDictIdsMV(_valueBlock);
@@ -194,6 +211,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public int[][] getIntValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.INT, false);
return _transformFunction.transformToIntValuesMV(_valueBlock);
@@ -202,6 +220,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public long[][] getLongValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.LONG, false);
return _transformFunction.transformToLongValuesMV(_valueBlock);
@@ -210,6 +229,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public float[][] getFloatValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.FLOAT, false);
return _transformFunction.transformToFloatValuesMV(_valueBlock);
@@ -218,6 +238,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public double[][] getDoubleValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.DOUBLE, false);
return _transformFunction.transformToDoubleValuesMV(_valueBlock);
@@ -226,6 +247,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public BigDecimal[][] getBigDecimalValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.BIG_DECIMAL, false);
return _transformFunction.transformToBigDecimalValuesMV(_valueBlock);
@@ -234,6 +256,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public String[][] getStringValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.STRING, false);
return _transformFunction.transformToStringValuesMV(_valueBlock);
@@ -242,6 +265,7 @@ public class TransformBlockValSet implements BlockValSet {
@Override
public byte[][][] getBytesValuesMV() {
+ checkTermination();
try (InvocationScope scope =
Tracing.getTracer().createScope(TransformBlockValSet.class)) {
recordTransformValues(scope, DataType.BYTES, false);
return _transformFunction.transformToBytesValuesMV(_valueBlock);
@@ -311,6 +335,11 @@ public class TransformBlockValSet implements BlockValSet {
}
}
+ /// Throws if the query is terminated or timed out. Called before each
evaluation of the transform function.
+ private void checkTermination() {
+ QueryThreadContext.checkTermination(_terminationScope);
+ }
+
private void recordTransformValues(InvocationRecording recording, DataType
dataType, boolean singleValue) {
if (recording.isEnabled()) {
int numDocs = _valueBlock.getNumDocs();
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSetTerminationTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSetTerminationTest.java
new file mode 100644
index 00000000000..94389522e30
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/docvalsets/TransformBlockValSetTerminationTest.java
@@ -0,0 +1,115 @@
+/**
+ * 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.pinot.core.operator.docvalsets;
+
+import org.apache.pinot.core.operator.blocks.ValueBlock;
+import org.apache.pinot.core.operator.transform.function.TransformFunction;
+import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.exception.EarlyTerminationException;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.exception.TerminationException;
+import org.apache.pinot.spi.query.QueryExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.expectThrows;
+
+
+/// Checks that [TransformBlockValSet] does not evaluate the transform
function once the query is terminated, timed
+/// out or interrupted.
+public class TransformBlockValSetTerminationTest {
+ private static final String[] VALUES = {"a", "b"};
+
+ @Test
+ public void testEvaluatesWhenQueryIsActive() {
+ TransformFunction transformFunction = mockTransformFunction();
+ try (QueryThreadContext ignore = QueryThreadContext.openForSseTest()) {
+ assertSame(newValSet(transformFunction).getStringValuesSV(), VALUES);
+ }
+ }
+
+ @Test
+ public void testEvaluatesWithoutQueryThreadContext() {
+ TransformFunction transformFunction = mockTransformFunction();
+ assertSame(newValSet(transformFunction).getStringValuesSV(), VALUES);
+ }
+
+ @Test
+ public void testStopsWhenQueryIsTerminated() {
+ TransformFunction transformFunction = mockTransformFunction();
+ QueryExecutionContext executionContext =
QueryExecutionContext.forSseTest();
+ try (QueryThreadContext ignore = QueryThreadContext.open(executionContext,
+ ThreadAccountantUtils.getNoOpAccountant())) {
+ executionContext.terminate(QueryErrorCode.QUERY_CANCELLATION,
"terminated for test");
+ TerminationException e =
+ expectThrows(TerminationException.class, () ->
newValSet(transformFunction).getStringValuesSV());
+ assertEquals(e.getErrorCode(), QueryErrorCode.QUERY_CANCELLATION);
+ }
+ verify(transformFunction, never()).transformToStringValuesSV(any());
+ }
+
+ @Test
+ public void testStopsWhenQueryIsTimedOut() {
+ TransformFunction transformFunction = mockTransformFunction();
+ long nowMs = System.currentTimeMillis();
+ QueryExecutionContext executionContext = new
QueryExecutionContext(QueryExecutionContext.QueryType.SSE, 123L,
+ "cid", "default", nowMs - 1000, nowMs - 1, nowMs - 1, "brokerId",
"instanceId", "");
+ try (QueryThreadContext ignore = QueryThreadContext.open(executionContext,
+ ThreadAccountantUtils.getNoOpAccountant())) {
+ QueryException e = expectThrows(QueryException.class, () ->
newValSet(transformFunction).getStringValuesSV());
+ assertEquals(e.getErrorCode(), QueryErrorCode.EXECUTION_TIMEOUT);
+ }
+ verify(transformFunction, never()).transformToStringValuesSV(any());
+ }
+
+ @Test
+ public void testStopsWhenThreadIsInterrupted() {
+ TransformFunction transformFunction = mockTransformFunction();
+ try (QueryThreadContext ignore = QueryThreadContext.openForSseTest()) {
+ Thread.currentThread().interrupt();
+ assertThrows(EarlyTerminationException.class, () ->
newValSet(transformFunction).getStringValuesSV());
+ } finally {
+ // Clear the flag in case the check did not consume it.
+ Thread.interrupted();
+ }
+ verify(transformFunction, never()).transformToStringValuesSV(any());
+ }
+
+ private static TransformFunction mockTransformFunction() {
+ TransformFunction transformFunction = mock(TransformFunction.class);
+ when(transformFunction.getName()).thenReturn("testTransform");
+
when(transformFunction.transformToStringValuesSV(any())).thenReturn(VALUES);
+ return transformFunction;
+ }
+
+ private static TransformBlockValSet newValSet(TransformFunction
transformFunction) {
+ ValueBlock valueBlock = mock(ValueBlock.class);
+ when(valueBlock.getNumDocs()).thenReturn(VALUES.length);
+ return new TransformBlockValSet(valueBlock, transformFunction);
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/TransformEvaluationDeadlineQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/TransformEvaluationDeadlineQueriesTest.java
new file mode 100644
index 00000000000..a29eff1350e
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/TransformEvaluationDeadlineQueriesTest.java
@@ -0,0 +1,160 @@
+/**
+ * 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.pinot.queries;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.function.FunctionUtils;
+import org.apache.pinot.core.operator.BaseOperator;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.query.QueryExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Checks that the query deadline stops the group-by processing of a block
between two transform evaluations.
+///
+/// The segment has one block. `DefaultGroupByExecutor.process()` handles the
whole block in one call. Each aggregation
+/// evaluates `sleep(col) + k` on the whole block. `col` is `SLEEP_MS` in
every row, so one evaluation takes
+/// `NUM_RECORDS * SLEEP_MS` = 200 ms, and the 10 aggregations take 2 s in
total. No termination check runs inside one
+/// block, except the check that `TransformBlockValSet` runs before each
transform evaluation. So with a 500 ms
+/// deadline, the query must fail with a timeout from inside the block after
about 600 ms. Without that check, the
+/// block runs to the end (2 s), and the timeout only comes from the
`nextBlock()` check at the next block boundary.
+public class TransformEvaluationDeadlineQueriesTest extends BaseQueriesTest {
+ private static final File INDEX_DIR =
+ new File(FileUtils.getTempDirectory(),
"TransformEvaluationDeadlineQueriesTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final String COLUMN = "col";
+ private static final int NUM_RECORDS = 20;
+ private static final int SLEEP_MS = 10;
+ private static final int NUM_AGGREGATIONS = 10;
+ // The first transform evaluation must start before the deadline. The margin
covers class loading and GC pauses on
+ // slow CI runners.
+ private static final long DEADLINE_MS = 500;
+ // The full block takes NUM_AGGREGATIONS * NUM_RECORDS * SLEEP_MS = 2 s.
Stopping at the deadline plus at most one
+ // more evaluation (200 ms) is well below this limit.
+ private static final long MAX_ELAPSED_MS = 1500;
+
+ private static final Schema SCHEMA = new
Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension(COLUMN, DataType.LONG)
+ .build();
+ private static final TableConfig TABLE_CONFIG =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+
+ private IndexSegment _indexSegment;
+
+ @Override
+ protected String getFilter() {
+ return "";
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return List.of(_indexSegment);
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteQuietly(INDEX_DIR);
+ List<GenericRow> records = new ArrayList<>(NUM_RECORDS);
+ for (int i = 0; i < NUM_RECORDS; i++) {
+ GenericRow record = new GenericRow();
+ // Every row sleeps SLEEP_MS: the aggregations call sleep(col).
+ record.putValue(COLUMN, (long) SLEEP_MS);
+ records.add(record);
+ }
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ segmentGeneratorConfig.setSegmentName(SEGMENT_NAME);
+ segmentGeneratorConfig.setOutDir(INDEX_DIR.getPath());
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(records));
+ driver.build();
+ ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(new
File(INDEX_DIR, SEGMENT_NAME), ReadMode.mmap);
+ _indexSegment = immutableSegment;
+ }
+
+ @Test
+ public void testDeadlineStopsBlockBetweenTransformEvaluations() {
+ if (!FunctionUtils.isAssertEnabled()) {
+ throw new SkipException("sleep() only sleeps when assertions are
enabled");
+ }
+ StringBuilder query = new StringBuilder("SELECT col");
+ for (int k = 0; k < NUM_AGGREGATIONS; k++) {
+ // A different constant per aggregation, so that each one is a different
transform expression. The constant is
+ // outside sleep(), so that every evaluation sleeps the same time.
+ query.append(", SUM(sleep(col) + ").append(k).append(")");
+ }
+ query.append(" FROM testTable GROUP BY col");
+
+ // Plan before the deadline starts: planning the first query of the JVM
can take longer than the deadline.
+ BaseOperator<?> operator = getOperator(query.toString());
+ long startMs = System.currentTimeMillis();
+ QueryExecutionContext executionContext =
+ new QueryExecutionContext(QueryExecutionContext.QueryType.SSE, 123L,
"cid", "default", startMs,
+ startMs + DEADLINE_MS, startMs + DEADLINE_MS, "brokerId",
"instanceId", "");
+ try (QueryThreadContext ignore = QueryThreadContext.open(executionContext,
+ ThreadAccountantUtils.getNoOpAccountant())) {
+ QueryException e = expectThrows(QueryException.class,
operator::nextBlock);
+ // The block must start before the deadline, so that the timeout comes
from inside the block.
+ assertTrue(e.getMessage().contains("transform function"),
e.getMessage());
+ assertEquals(e.getErrorCode(), QueryErrorCode.EXECUTION_TIMEOUT);
+ }
+ long elapsedMs = System.currentTimeMillis() - startMs;
+ assertTrue(elapsedMs < MAX_ELAPSED_MS,
+ "The query must stop soon after the deadline (" + DEADLINE_MS + " ms),
but it ran " + elapsedMs + " ms");
+ }
+
+ @AfterClass
+ public void tearDown() {
+ _indexSegment.destroy();
+ FileUtils.deleteQuietly(INDEX_DIR);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
