This is an automated email from the ASF dual-hosted git repository.
ankitsultana 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 743405ee16 [timeseries] Support Raw Table Names for Time Series
Queries (#16003)
743405ee16 is described below
commit 743405ee16f67a7e6dba3125f9ea74db2582f826
Author: Shaurya Chaturvedi <[email protected]>
AuthorDate: Thu Jun 5 13:06:15 2025 -0700
[timeseries] Support Raw Table Names for Time Series Queries (#16003)
---
.../tests/TimeSeriesIntegrationTest.java | 12 +++++
pinot-timeseries/pinot-timeseries-planner/pom.xml | 4 ++
.../tsdb/planner/TimeSeriesQueryEnvironment.java | 7 ++-
.../tsdb/planner/physical/TableScanVisitor.java | 56 +++++++++++++++++++++-
.../tsdb/spi/plan/LeafTimeSeriesPlanNode.java | 5 ++
5 files changed, 82 insertions(+), 2 deletions(-)
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesIntegrationTest.java
index c6e21bc486..58ac3fcc0f 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesIntegrationTest.java
@@ -152,6 +152,18 @@ public class TimeSeriesIntegrationTest extends
BaseClusterIntegrationTest {
);
}
+ @Test
+ public void testTableWithoutType() {
+ String query = String.format(
+
"fetch{table=\"mytable\",filter=\"\",ts_column=\"%s\",ts_unit=\"MILLISECONDS\",value=\"%s\"}"
+ + " | max{%s} | transformNull{0} | keepLastValue{}",
+ TS_COLUMN, TOTAL_TRIPS_COLUMN, DEVICE_OS_COLUMN
+ );
+ runGroupedTimeSeriesQuery(query, 3, (ts, val, row) ->
+ assertEquals(val, ts <= DATA_START_TIME_SEC ? 0L : VIEWS_MAX_VALUE)
+ );
+ }
+
private void runGroupedTimeSeriesQuery(String query, int expectedGroups,
TimeSeriesValidator validator) {
JsonNode result = getTimeseriesQuery(query, QUERY_START_TIME_SEC,
QUERY_END_TIME_SEC);
System.out.println(result);
diff --git a/pinot-timeseries/pinot-timeseries-planner/pom.xml
b/pinot-timeseries/pinot-timeseries-planner/pom.xml
index 1c7e6c6144..74aa3c1ffd 100644
--- a/pinot-timeseries/pinot-timeseries-planner/pom.xml
+++ b/pinot-timeseries/pinot-timeseries-planner/pom.xml
@@ -52,6 +52,10 @@
<artifactId>testng</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-query-planner</artifactId>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/TimeSeriesQueryEnvironment.java
b/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/TimeSeriesQueryEnvironment.java
index 5265f61747..e4a72e4b51 100644
---
a/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/TimeSeriesQueryEnvironment.java
+++
b/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/TimeSeriesQueryEnvironment.java
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.query.routing.table.ImplicitHybridTableRouteProvider;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.trace.RequestContext;
import org.apache.pinot.tsdb.planner.physical.TableScanVisitor;
@@ -75,7 +76,8 @@ public class TimeSeriesQueryEnvironment {
throw new RuntimeException("Failed to instantiate logical planner for
language: " + language, e);
}
}
- TableScanVisitor.INSTANCE.init(_routingManager);
+ // TODO(timeseries): Add support for logical tables in the future.
+ TableScanVisitor.INSTANCE.init(_routingManager, new
ImplicitHybridTableRouteProvider(), _tableCache);
}
public TimeSeriesLogicalPlanResult buildLogicalPlan(RangeTimeSeriesRequest
request) {
@@ -87,6 +89,9 @@ public class TimeSeriesQueryEnvironment {
public TimeSeriesDispatchablePlan buildPhysicalPlan(RangeTimeSeriesRequest
timeSeriesRequest,
RequestContext requestContext, TimeSeriesLogicalPlanResult logicalPlan) {
+ // Step-0: Add table type info to the logical plan.
+ logicalPlan = new
TimeSeriesLogicalPlanResult(TableScanVisitor.INSTANCE.addTableTypeInfoToPlan(
+ logicalPlan.getPlanNode()), logicalPlan.getTimeBuckets());
// Step-1: Assign segments to servers for each leaf node.
TableScanVisitor.Context scanVisitorContext =
TableScanVisitor.createContext(requestContext.getRequestId());
TableScanVisitor.INSTANCE.assignSegmentsToPlan(logicalPlan.getPlanNode(),
logicalPlan.getTimeBuckets(),
diff --git
a/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/physical/TableScanVisitor.java
b/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/physical/TableScanVisitor.java
index 2217f41570..584ce13962 100644
---
a/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/physical/TableScanVisitor.java
+++
b/pinot-timeseries/pinot-timeseries-planner/src/main/java/org/apache/pinot/tsdb/planner/physical/TableScanVisitor.java
@@ -19,10 +19,13 @@
package org.apache.pinot.tsdb.planner.physical;
import com.google.common.base.Preconditions;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.request.DataSource;
import org.apache.pinot.common.request.Expression;
@@ -31,6 +34,8 @@ import org.apache.pinot.common.request.QuerySource;
import org.apache.pinot.core.routing.RoutingManager;
import org.apache.pinot.core.routing.RoutingTable;
import org.apache.pinot.core.transport.ServerInstance;
+import org.apache.pinot.core.transport.TableRouteInfo;
+import org.apache.pinot.query.routing.table.TableRouteProvider;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.tsdb.spi.TimeBuckets;
import org.apache.pinot.tsdb.spi.plan.BaseTimeSeriesPlanNode;
@@ -40,12 +45,16 @@ import
org.apache.pinot.tsdb.spi.plan.LeafTimeSeriesPlanNode;
public class TableScanVisitor {
public static final TableScanVisitor INSTANCE = new TableScanVisitor();
private RoutingManager _routingManager;
+ private TableRouteProvider _tableRouteProvider;
+ private TableCache _tableCache;
private TableScanVisitor() {
}
- public void init(RoutingManager routingManager) {
+ public void init(RoutingManager routingManager, TableRouteProvider
tableRouteProvider, TableCache tableCache) {
_routingManager = routingManager;
+ _tableRouteProvider = tableRouteProvider;
+ _tableCache = tableCache;
}
public void assignSegmentsToPlan(BaseTimeSeriesPlanNode planNode,
TimeBuckets timeBuckets, Context context) {
@@ -68,6 +77,51 @@ public class TableScanVisitor {
}
}
+ /**
+ * Adds table type information (offline/realtime) to the plan node.
+ * If the plan node is a leaf node, it retrieves the table route info and
updates the table name with type.
+ * If the plan node has child nodes, it recursively processes each child
node.
+ *
+ * @param planNode The {@link BaseTimeSeriesPlanNode} to process.
+ * @return The updated {@link BaseTimeSeriesPlanNode} with table type
information.
+ */
+ public BaseTimeSeriesPlanNode addTableTypeInfoToPlan(BaseTimeSeriesPlanNode
planNode) {
+ if (planNode instanceof LeafTimeSeriesPlanNode) {
+ LeafTimeSeriesPlanNode sfpNode = (LeafTimeSeriesPlanNode) planNode;
+ TableRouteInfo routeInfo =
_tableRouteProvider.getTableRouteInfo(sfpNode.getTableName(), _tableCache,
+ _routingManager);
+ String tableNameWithType = getTableNameWithType(routeInfo);
+ Preconditions.checkNotNull(tableNameWithType, "Table not found for table
name: " + sfpNode.getTableName());
+ return sfpNode.withTableName(tableNameWithType);
+ }
+
+ List<BaseTimeSeriesPlanNode> newInputs = new ArrayList<>();
+ for (BaseTimeSeriesPlanNode childNode : planNode.getInputs()) {
+ newInputs.add(addTableTypeInfoToPlan(childNode));
+ }
+ return planNode.withInputs(newInputs);
+ }
+
+
+ /**
+ * Returns the table name with type (offline/realtime) if the table exists,
otherwise returns null.
+ *
+ * @param routeInfo The {@link TableRouteInfo} for the table.
+ * @return The table name with type, or null if the table does not exist.
+ */
+ @Nullable
+ private String getTableNameWithType(TableRouteInfo routeInfo) {
+ Preconditions.checkState(!routeInfo.isHybrid(),
+ "Hybrid tables are not supported yet for timeseries queries");
+ if (routeInfo.isOffline()) {
+ return routeInfo.getOfflineTableName();
+ }
+ if (routeInfo.isRealtime()) {
+ return routeInfo.getRealtimeTableName();
+ }
+ return null;
+ }
+
public static Context createContext(Long requestId) {
return new Context(requestId);
}
diff --git
a/pinot-timeseries/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java
b/pinot-timeseries/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java
index b0c7046466..d10aa433e1 100644
---
a/pinot-timeseries/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java
+++
b/pinot-timeseries/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/plan/LeafTimeSeriesPlanNode.java
@@ -76,6 +76,11 @@ public class LeafTimeSeriesPlanNode extends
BaseTimeSeriesPlanNode {
_filterExpression, _valueExpression, newAggInfo, _groupByExpressions,
_limit, _queryOptions);
}
+ public LeafTimeSeriesPlanNode withTableName(String newTableName) {
+ return new LeafTimeSeriesPlanNode(_id, _inputs, newTableName, _timeColumn,
_timeUnit, _offsetSeconds,
+ _filterExpression, _valueExpression, _aggInfo, _groupByExpressions,
_limit, _queryOptions);
+ }
+
@Override
public BaseTimeSeriesPlanNode withInputs(List<BaseTimeSeriesPlanNode>
newInputs) {
return new LeafTimeSeriesPlanNode(_id, newInputs, _tableName, _timeColumn,
_timeUnit, _offsetSeconds,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
