[timeseries] Minor Time Series Engine Improvements (#14227)

apache · Oct 15, 2024 · 47b322b · 47b322b
1 parent 68cb616
commit 47b322b
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 8 deletions.
diff --git a/...rc/main/java/org/apache/pinot/core/operator/timeseries/TimeSeriesAggregationOperator.java b/...rc/main/java/org/apache/pinot/core/operator/timeseries/TimeSeriesAggregationOperator.java
@@ -105,6 +105,9 @@ protected TimeSeriesResultsBlock getNextBlock() {
         case LONG:
           tagValues[i] = ArrayUtils.toObject(blockValSet.getLongValuesSV());
           break;
+        case INT:
+          tagValues[i] = ArrayUtils.toObject(blockValSet.getIntValuesSV());
+          break;
         default:
           throw new NotImplementedException("Can't handle types other than string and long");
       }

diff --git a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
@@ -253,7 +253,8 @@ public void processTimeSeriesQuery(String serializedPlan, Map<String, String> me
     final Consumer<Throwable> handleErrors = (t) -> {
       Map<String, String> errorMetadata = new HashMap<>();
       errorMetadata.put(WorkerResponseMetadataKeys.ERROR_TYPE, t.getClass().getSimpleName());
-      errorMetadata.put(WorkerResponseMetadataKeys.ERROR_MESSAGE, t.getMessage());
+      errorMetadata.put(WorkerResponseMetadataKeys.ERROR_MESSAGE, t.getMessage() == null
+          ? "Unknown error: no message" : t.getMessage());
       responseObserver.onNext(Worker.TimeSeriesResponse.newBuilder().putAllMetadata(errorMetadata).build());
       responseObserver.onCompleted();
     };
@@ -280,6 +281,7 @@ public void processTimeSeriesQuery(String serializedPlan, Map<String, String> me
         }
       });
     } catch (Throwable t) {
+      LOGGER.error("Error running time-series query", t);
       handleErrors.accept(t);
     }
   }

diff --git a/...est/java/org/apache/pinot/query/runtime/timeseries/PhysicalTimeSeriesPlanVisitorTest.java b/...est/java/org/apache/pinot/query/runtime/timeseries/PhysicalTimeSeriesPlanVisitorTest.java
@@ -54,7 +54,7 @@ public void testCompileQueryContext() {
       assertEquals(queryContext.getTimeSeriesContext().getTimeColumn(), timeColumn);
       assertEquals(queryContext.getTimeSeriesContext().getValueExpression().getIdentifier(), "orderCount");
       assertEquals(queryContext.getFilter().toString(),
-          "(cityName = 'Chicago' AND orderTime >= '1000' AND orderTime <= '2000')");
+          "(cityName = 'Chicago' AND orderTime >= '1000' AND orderTime < '2000')");
     }
     // Case-2: With offset, complex group-by expression, complex value, and non-empty filter
     {
@@ -75,7 +75,7 @@ public void testCompileQueryContext() {
       assertEquals(queryContext.getTimeSeriesContext().getValueExpression().toString(), "times(orderCount,'2')");
       assertNotNull(queryContext.getFilter());
       assertEquals(queryContext.getFilter().toString(),
-          "(cityName = 'Chicago' AND orderTime >= '990' AND orderTime <= '1990')");
+          "(cityName = 'Chicago' AND orderTime >= '990' AND orderTime < '1990')");
     }
   }
 }
diff --git a/...t-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java b/...t-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java
@@ -119,7 +119,7 @@ public String getEffectiveFilter(TimeBuckets timeBuckets) {
     long endTime =
         _timeUnit.convert(Duration.ofSeconds(
             timeBuckets.getEndTime() + timeBuckets.getBucketSize().toSeconds() - _offsetSeconds));
-    String addnFilter = String.format("%s >= %d AND %s <= %d", _timeColumn, startTime, _timeColumn, endTime);
+    String addnFilter = String.format("%s >= %d AND %s < %d", _timeColumn, startTime, _timeColumn, endTime);
     if (filter.strip().isEmpty()) {
       return addnFilter;
     }

diff --git a/...meseries-spi/src/test/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNodeTest.java b/...meseries-spi/src/test/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNodeTest.java
@@ -46,23 +46,23 @@ public void testGetEffectiveFilter() {
           new LeafTimeSeriesPlanNode(ID, Collections.emptyList(), TABLE, TIME_COLUMN, TIME_UNIT, 0L, "", "value_col",
               new AggInfo("SUM", null), Collections.singletonList("cityName"));
       assertEquals(planNode.getEffectiveFilter(timeBuckets),
-          "orderTime >= " + expectedStartTimeInFilter + " AND orderTime <= " + expectedEndTimeInFilter);
+          "orderTime >= " + expectedStartTimeInFilter + " AND orderTime < " + expectedEndTimeInFilter);
     }
     // Case-2: Offset, but empty filter
     {
       LeafTimeSeriesPlanNode planNode =
           new LeafTimeSeriesPlanNode(ID, Collections.emptyList(), TABLE, TIME_COLUMN, TIME_UNIT, 123L, "", "value_col",
               new AggInfo("SUM", null), Collections.singletonList("cityName"));
       assertEquals(planNode.getEffectiveFilter(timeBuckets),
-          "orderTime >= " + (expectedStartTimeInFilter - 123) + " AND orderTime <= " + (expectedEndTimeInFilter - 123));
+          "orderTime >= " + (expectedStartTimeInFilter - 123) + " AND orderTime < " + (expectedEndTimeInFilter - 123));
     }
     // Case-3: Offset and non-empty filter
     {
       LeafTimeSeriesPlanNode planNode =
           new LeafTimeSeriesPlanNode(ID, Collections.emptyList(), TABLE, TIME_COLUMN, TIME_UNIT, 123L, nonEmptyFilter,
               "value_col", new AggInfo("SUM", null), Collections.singletonList("cityName"));
       assertEquals(planNode.getEffectiveFilter(timeBuckets),
-          String.format("(%s) AND (orderTime >= %s AND orderTime <= %s)", nonEmptyFilter,
+          String.format("(%s) AND (orderTime >= %s AND orderTime < %s)", nonEmptyFilter,
               (expectedStartTimeInFilter - 123), (expectedEndTimeInFilter - 123)));
     }
     // Case-4: Offset, and non-empty filter, and time-unit that is not seconds
@@ -71,7 +71,7 @@ public void testGetEffectiveFilter() {
           new LeafTimeSeriesPlanNode(ID, Collections.emptyList(), TABLE, TIME_COLUMN, TimeUnit.MILLISECONDS, 123L,
               nonEmptyFilter, "value_col", new AggInfo("SUM", null), Collections.singletonList("cityName"));
       assertEquals(planNode.getEffectiveFilter(timeBuckets),
-          String.format("(%s) AND (orderTime >= %s AND orderTime <= %s)", nonEmptyFilter,
+          String.format("(%s) AND (orderTime >= %s AND orderTime < %s)", nonEmptyFilter,
               (expectedStartTimeInFilter * 1000 - 123 * 1000), (expectedEndTimeInFilter * 1000 - 123 * 1000)));
     }
   }