Throw an exception when MV columns are present in the order-by expression list in selection order-by only queries

compatibility-verifier/sample-test-suite/config/queries/feature-test-2-sql-realtime.queries

@@ -18,7 +18,7 @@
 #
 
 # Selection
-SELECT longDimSV1, intDimMV1 FROM FeatureTest2 WHERE generationNumber = __GENERATION_NUMBER__ AND (stringDimSV1 != 's1-6' AND longDimSV1 BETWEEN 10 AND 1000 OR (intDimMV1 < 42 AND stringDimMV2 IN ('m2-0-0', 'm2-2-0') AND intDimMV2 NOT IN (6,72))) ORDER BY longDimSV1, intDimMV1
+SELECT longDimSV1, intDimMV1 FROM FeatureTest2 WHERE generationNumber = __GENERATION_NUMBER__ AND (stringDimSV1 != 's1-6' AND longDimSV1 BETWEEN 10 AND 1000 OR (intDimMV1 < 42 AND stringDimMV2 IN ('m2-0-0', 'm2-2-0') AND intDimMV2 NOT IN (6,72))) ORDER BY longDimSV1
 # This is a subset comparison query: there are 10 qualified records in total, the query limit the number of returned records as 5, the test compares the returned results to be a subset of total qualified results.
 # To generate the result file, use this query instead:
 # SELECT longDimSV1, intDimMV1 FROM FeatureTest2 WHERE generationNumber = __GENERATION_NUMBER__ AND (stringDimSV1 != 's1-6' AND longDimSV1 BETWEEN 10 AND 1000 OR (intDimMV1 < 42 AND stringDimMV2 IN ('m2-0-0', 'm2-2-0') AND intDimMV2 NOT IN (6,72)))

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java

@@ -73,7 +73,6 @@ public enum ServerMeter implements AbstractMetrics.Meter {
   NUM_RESIZES("numResizes", false),
   NO_TABLE_ACCESS("tables", true),
   INDEXING_FAILURES("attributeValues", true),
-  QUERY_HAS_MV_SELECTION_ORDER_BY("queryHasMVSelectionOrderBy", false),
 
   READINESS_CHECK_OK_CALLS("readinessCheck", true),
   READINESS_CHECK_BAD_CALLS("readinessCheck", true),

pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java

@@ -21,7 +21,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import org.apache.pinot.common.metrics.ServerMeter;
 import org.apache.pinot.common.metrics.ServerMetrics;
 import org.apache.pinot.common.utils.DataTable.MetadataKey;
 import org.apache.pinot.core.common.Operator;
@@ -32,13 +31,10 @@
 import org.apache.pinot.core.query.request.context.ThreadTimer;
 import org.apache.pinot.segment.spi.FetchContext;
 import org.apache.pinot.segment.spi.IndexSegment;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 public class InstanceResponseOperator extends BaseOperator<InstanceResponseBlock> {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(InstanceResponseOperator.class);
   private static final String EXPLAIN_NAME = "INSTANCE_RESPONSE";
 
   private final BaseCombineOperator _combineOperator;
@@ -85,15 +81,12 @@ public static long calSystemActivitiesCpuTimeNs(long totalWallClockTimeNs, long
 
   @Override
   protected InstanceResponseBlock getNextBlock() {
-    IntermediateResultsBlock intermediateResultsBlock;
-    InstanceResponseBlock instanceResponseBlock;
-
     if (ThreadTimer.isThreadCpuTimeMeasurementEnabled()) {
       long startWallClockTimeNs = System.nanoTime();
 
       ThreadTimer mainThreadTimer = new ThreadTimer();
-      intermediateResultsBlock = getCombinedResults();
-      instanceResponseBlock = new InstanceResponseBlock(intermediateResultsBlock);
+      IntermediateResultsBlock intermediateResultsBlock = getCombinedResults();
+      InstanceResponseBlock instanceResponseBlock = new InstanceResponseBlock(intermediateResultsBlock);
       long mainThreadCpuTimeNs = mainThreadTimer.getThreadTimeNs();
 
       long totalWallClockTimeNs = System.nanoTime() - startWallClockTimeNs;
@@ -113,28 +106,11 @@ protected InstanceResponseBlock getNextBlock() {
       responseMetaData.put(MetadataKey.THREAD_CPU_TIME_NS.getName(), String.valueOf(threadCpuTimeNs));
       responseMetaData
           .put(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName(), String.valueOf(systemActivitiesCpuTimeNs));
-    } else {
-      intermediateResultsBlock = getCombinedResults();
-      instanceResponseBlock = new InstanceResponseBlock(intermediateResultsBlock);
-    }
-
-    // TODO: Remove this once the SelectionOrderByOperator is modified to throw an exception to catch cases where
-    //       an MV column (identifier or via transform) is present on the order-by list
-    logAndEmitMetricForQueryHasMVSelectionOrderBy(intermediateResultsBlock);
-    return instanceResponseBlock;
-  }
-
-  private void logAndEmitMetricForQueryHasMVSelectionOrderBy(IntermediateResultsBlock intermediateResultsBlock) {
-    if (!intermediateResultsBlock.isQueryHasMVSelectionOrderBy()) {
-      return;
-    }
 
-    String tableName = _queryContext.getTableName();
-    if (_serverMetrics != null) {
-      _serverMetrics.addMeteredTableValue(tableName, ServerMeter.QUERY_HAS_MV_SELECTION_ORDER_BY, 1);
+      return instanceResponseBlock;
+    } else {
+      return new InstanceResponseBlock(getCombinedResults());
     }
-    LOGGER.warn("Table {} has MV column in ORDER BY. Expressions: {}", tableName,
-        _queryContext.getOrderByExpressions());
   }
 
   private IntermediateResultsBlock getCombinedResults() {

pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java

@@ -82,7 +82,6 @@ public class IntermediateResultsBlock implements Block {
   private long _resizeTimeMs;
   private long _executionThreadCpuTimeNs;
   private int _numServerThreads;
-  private boolean _queryHasMVSelectionOrderBy;
 
   private Table _table;
 
@@ -349,14 +348,6 @@ public void setNumGroupsLimitReached(boolean numGroupsLimitReached) {
     _numGroupsLimitReached = numGroupsLimitReached;
   }
 
-  public boolean isQueryHasMVSelectionOrderBy() {
-    return _queryHasMVSelectionOrderBy;
-  }
-
-  public void setQueryHasMVSelectionOrderBy(boolean queryHasMVSelectionOrderBy) {
-    _queryHasMVSelectionOrderBy = queryHasMVSelectionOrderBy;
-  }
-
   public DataTable getDataTable()
       throws Exception {
 

pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOrderByOperator.java

@@ -47,6 +47,7 @@
 import org.apache.pinot.segment.spi.IndexSegment;
 import org.apache.pinot.segment.spi.datasource.DataSource;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
 import org.apache.pinot.spi.utils.ByteArray;
 import org.roaringbitmap.RoaringBitmap;
 
@@ -87,8 +88,6 @@ public class SelectionOrderByOperator extends BaseOperator<IntermediateResultsBl
   private int _numDocsScanned = 0;
   private long _numEntriesScannedPostFilter = 0;
 
-  private boolean _queryHasMVSelectionOrderBy = false;
-
   public SelectionOrderByOperator(IndexSegment indexSegment, QueryContext queryContext,
       List<ExpressionContext> expressions, TransformOperator transformOperator, boolean allOrderByColsPreSorted) {
     _indexSegment = indexSegment;
@@ -131,7 +130,10 @@ private Comparator<Object[]> getComparator() {
       if (_orderByExpressionMetadata[i].isSingleValue()) {
         valueIndexList.add(i);
       } else {
-        _queryHasMVSelectionOrderBy = true;
+        // MV columns should not be part of the selection order by only list
+        throw new BadQueryRequestException(
+            String.format("MV expression: %s should not be included in the ORDER-BY clause",
+                _orderByExpressions.get(i)));
       }
     }
 
@@ -240,16 +242,13 @@ private Comparator<Object[]> getComparator() {
 
   @Override
   protected IntermediateResultsBlock getNextBlock() {
-    IntermediateResultsBlock resultsBlock;
     if (_allOrderByColsPreSorted) {
-      resultsBlock = computeAllPreSorted();
+      return computeAllPreSorted();
     } else if (_expressions.size() == _orderByExpressions.size()) {
-      resultsBlock = computeAllOrdered();
+      return computeAllOrdered();
     } else {
-      resultsBlock = computePartiallyOrdered();
+      return computePartiallyOrdered();
     }
-    resultsBlock.setQueryHasMVSelectionOrderBy(_queryHasMVSelectionOrderBy);
-    return resultsBlock;
   }
 
   private IntermediateResultsBlock computeAllPreSorted() {

pinot-core/src/test/java/org/apache/pinot/queries/MultiValueRawQueriesTest.java

@@ -42,6 +42,7 @@
 import org.apache.pinot.spi.data.FieldSpec;
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
 import org.apache.pinot.spi.utils.ReadMode;
 import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
 import org.testng.annotations.AfterClass;
@@ -51,6 +52,7 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
 import static org.testng.Assert.assertTrue;
 
 
@@ -386,15 +388,72 @@ public void testSelectQueries() {
     }
   }
 
+  /**
+   * Today selection ORDER BY only on MV columns (irrespective of whether it's dictionary based or raw) doesn't work
+   * as the semantics of how such queries should behave isn't clear. Such queries should always fail.
+   */
   @Test
-  public void testNonAggregateMVGroupBY() {
+  public void testSelectionOrderBy() {
     {
-      // TODO: Today selection ORDER BY only on MV columns (irrespective of whether it's dictionary based or raw)
-      //       doesn't work. Fix ORDER BY only for MV columns
       String query = "SELECT mvFloatCol from testTable WHERE mvFloatCol < 5 ORDER BY mvFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query));
+
+      String query1 = "SELECT mvRawFloatCol from testTable WHERE mvRawFloatCol < 5 ORDER BY mvRawFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query1));
+
+      String query2 = "SELECT mvIntCol, mvFloatCol from testTable ORDER BY mvIntCol, mvFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query2));
+
+      String query3 = "SELECT mvRawIntCol, mvRawFloatCol from testTable ORDER BY mvRawIntCol, mvRawFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query3));
+    }
+    {
+      String query = "SELECT mvFloatCol, svIntCol from testTable WHERE mvFloatCol < 5 ORDER BY mvFloatCol, svIntCol "
+          + "LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query));
+
+      String query1 = "SELECT mvRawFloatCol, svIntCol from testTable WHERE mvRawFloatCol < 5 ORDER BY mvRawFloatCol, "
+          + "svIntCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query1));
+    }
+    {
+      String query = "SELECT svIntCol, mvFloatCol from testTable WHERE mvRawFloatCol < 5 ORDER BY svIntCol, "
+          + "mvFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query));
+
+      String query1 = "SELECT svIntCol, mvRawFloatCol from testTable WHERE mvRawFloatCol < 5 ORDER BY svIntCol, "
+          + "mvRawFloatCol LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query1));
+    }
+    {
+      String query = "SELECT VALUEIN(mvIntCol, '0') from testTable WHERE mvIntCol IN (0) ORDER BY "
+          + "VALUEIN(mvIntCol, '0') DESC LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query));
+
+      String query1 = "SELECT VALUEIN(mvRawIntCol, '0') from testTable WHERE mvRawIntCol IN (0) ORDER BY "
+          + "VALUEIN(mvRawIntCol, '0') DESC LIMIT 10";
+      assertThrows(BadQueryRequestException.class, () -> getBrokerResponse(query1));
+    }
+    {
+      // Arraylength eventually translates to a SV column, so this should pass
+      String query = "SELECT ARRAYLENGTH(mvRawLongCol), ARRAYLENGTH(mvLongCol) from testTable ORDER BY "
+          + "ARRAYLENGTH(mvRawLongCol), ARRAYLENGTH(mvLongCol) LIMIT 10";
       BrokerResponseNative brokerResponseNative = getBrokerResponse(query);
-      assertEquals(brokerResponseNative.getProcessingExceptions().size(), 2);
+      assertTrue(brokerResponseNative.getProcessingExceptions() == null
+          || brokerResponseNative.getProcessingExceptions().size() == 0);
+      ResultTable resultTable = brokerResponseNative.getResultTable();
+      assertEquals(resultTable.getRows().size(), 10);
+      List<Object[]> recordRows = resultTable.getRows();
+      for (Object[] row : recordRows) {
+        assertEquals(row.length, 2);
+        assertEquals(row[0], MV_LENGTH);
+        assertEquals(row[1], MV_LENGTH);
+      }
     }
+  }
+
+  @Test
+  public void testNonAggregateMVGroupBy() {
     {
       // Test a group by query on some raw MV rows. Order by on SV column added for determinism
       String query = "SELECT svIntCol, mvRawFloatCol, mvRawDoubleCol, mvRawStringCol from testTable GROUP BY "