You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by fr...@apache.org on 2026/09/11 06:29:36 UTC
(druid) 07/09: feat(sql): add native sys.server_segments support
This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch codex/native-sys-segments
in repository https://gitbox.apache.org/repos/asf/druid.git
commit 414bd487d449da3bbacf3ef046972fa603653faa
Author: Frank Chen <fr...@outlook.com>
AuthorDate: Fri Sep 4 12:00:21 2026 +0800
feat(sql): add native sys.server_segments support
---
.../query/NativeSysServerSegmentsQueryTest.java | 77 ++++++++++
.../server/system/module/SystemTableModule.java | 3 +
.../table/ServerSegmentsTableDataProvider.java | 103 ++++++++++++++
.../table/ServerSegmentsTableDescriptor.java | 89 ++++++++++++
.../system/table/SystemTableDataProvider.java | 4 +-
.../ServerSegmentsTableDataProviderTest.java | 157 +++++++++++++++++++++
.../main/java/org/apache/druid/cli/CliBroker.java | 6 +
.../calcite/schema/NativeServerSegmentsTable.java | 72 ++++++++++
.../druid/sql/calcite/schema/SystemSchema.java | 21 +--
.../druid/sql/calcite/schema/SystemSchemaTest.java | 3 +-
.../schema/SystemTableDataProviderTest.java | 11 ++
11 files changed, 533 insertions(+), 13 deletions(-)
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServerSegmentsQueryTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServerSegmentsQueryTest.java
new file mode 100644
index 00000000000..51d8843fa76
--- /dev/null
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/NativeSysServerSegmentsQueryTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.druid.testing.embedded.query;
+
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.run.NativeSqlEngine;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Map;
+
+public class NativeSysServerSegmentsQueryTest extends EmbeddedClusterTestBase
+{
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addServer(new EmbeddedCoordinator())
+ .addServer(new EmbeddedOverlord())
+ .addServer(new EmbeddedHistorical())
+ .addServer(new EmbeddedBroker());
+ }
+
+ @ParameterizedTest(name = "plannerStrategy = {0}")
+ @ValueSource(strings = {
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED,
+ QueryContexts.NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED
+ })
+ public void testNativeAggregationUsesBrokerLocalProvider(final String plannerStrategy)
+ {
+ final String result = cluster.runSql(
+ "SELECT COUNT(*), COUNT(DISTINCT server), COUNT(DISTINCT segment_id) "
+ + "FROM sys.server_segments",
+ nativeQueryContext(plannerStrategy)
+ );
+
+ Assertions.assertEquals("0,0,0", result);
+ }
+
+ private static Map<String, Object> nativeQueryContext(final String plannerStrategy)
+ {
+ return Map.of(
+ QueryContexts.ENGINE,
+ NativeSqlEngine.NAME,
+ PlannerContext.CTX_USE_NATIVE_QUERY_FOR_SYSTEM_TABLES,
+ true,
+ QueryContexts.CTX_NATIVE_QUERY_SQL_PLANNING_MODE,
+ plannerStrategy
+ );
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java b/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java
index c31cf51da86..cdf58caa14d 100644
--- a/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java
+++ b/server/src/main/java/org/apache/druid/server/system/module/SystemTableModule.java
@@ -28,6 +28,7 @@ import org.apache.druid.query.SystemTableDataSource;
import org.apache.druid.server.system.table.SegmentsTableDescriptor;
import org.apache.druid.server.system.table.ServerPropertiesTableDataProvider;
import org.apache.druid.server.system.table.ServerPropertiesTableDescriptor;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
import org.apache.druid.server.system.table.SystemTableDataProvider;
import org.apache.druid.server.system.table.SystemTableDescriptor;
@@ -49,6 +50,8 @@ public class SystemTableModule implements Module
final MapBinder<String, SystemTableDescriptor> descriptorBinder = MapBinder.newMapBinder(binder, String.class, SystemTableDescriptor.class);
descriptorBinder.addBinding(ServerPropertiesTableDescriptor.TABLE_NAME)
.toInstance(new ServerPropertiesTableDescriptor());
+ descriptorBinder.addBinding(ServerSegmentsTableDescriptor.TABLE_NAME)
+ .toInstance(new ServerSegmentsTableDescriptor());
descriptorBinder.addBinding(SegmentsTableDescriptor.TABLE_NAME)
.toInstance(new SegmentsTableDescriptor());
diff --git a/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDataProvider.java b/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDataProvider.java
new file mode 100644
index 00000000000..6150c395b8c
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDataProvider.java
@@ -0,0 +1,103 @@
+/*
+ * 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.druid.server.system.table;
+
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import org.apache.druid.client.ImmutableDruidServer;
+import org.apache.druid.client.TimelineServerView;
+import org.apache.druid.query.BatchedInlineDataSource;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizationUtils;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.timeline.DataSegment;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/** Native row supplier for {@code sys.server_segments}. */
+public class ServerSegmentsTableDataProvider implements SystemTableDataProvider
+{
+ private final TimelineServerView serverView;
+ private final AuthorizerMapper authorizerMapper;
+
+ @Inject
+ public ServerSegmentsTableDataProvider(
+ final TimelineServerView serverView,
+ final AuthorizerMapper authorizerMapper
+ )
+ {
+ this.serverView = serverView;
+ this.authorizerMapper = authorizerMapper;
+ }
+
+ @Override
+ public Optional<DataSource> getAuthorizedDataSource(
+ final SystemTableQueryRequest request,
+ final Iterable<Object[]> authorizedRows
+ )
+ {
+ final int[] projects = request.columns()
+ .stream()
+ .mapToInt(ServerSegmentsTableDescriptor.ROW_SIGNATURE::indexOf)
+ .toArray();
+ return Optional.of(
+ new BatchedInlineDataSource(
+ Iterables.transform(authorizedRows, row -> projectRow(row, projects)),
+ request.rowSignature()
+ )
+ );
+ }
+
+ @Override
+ public Iterable<Object[]> getRows(
+ final List<DimFilter> filters,
+ final AuthenticationResult authenticationResult
+ )
+ {
+ final Iterable<Iterable<Object[]>> rowsByServer = Iterables.transform(
+ serverView.getDruidServers(),
+ server -> getAuthorizedRows(server, authenticationResult)
+ );
+ return Iterables.concat(rowsByServer);
+ }
+
+ private Iterable<Object[]> getAuthorizedRows(
+ final ImmutableDruidServer server,
+ final AuthenticationResult authenticationResult
+ )
+ {
+ final Iterable<DataSegment> authorizedSegments = AuthorizationUtils.filterAuthorizedResources(
+ authenticationResult,
+ server.iterateAllSegments(),
+ segment -> Collections.singletonList(
+ AuthorizationUtils.DATASOURCE_READ_RA_GENERATOR.apply(segment.getDataSource())
+ ),
+ authorizerMapper
+ );
+ return Iterables.transform(
+ authorizedSegments,
+ segment -> new Object[]{server.getHost(), segment.getId().toString()}
+ );
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDescriptor.java b/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDescriptor.java
new file mode 100644
index 00000000000..e973590d494
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/system/table/ServerSegmentsTableDescriptor.java
@@ -0,0 +1,89 @@
+/*
+ * 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.druid.server.system.table;
+
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationResult;
+import org.apache.druid.server.security.AuthorizationUtils;
+import org.apache.druid.server.security.ForbiddenException;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+
+import java.util.Collections;
+import java.util.Set;
+
+/** Descriptor for the native {@code sys.server_segments} table. */
+public class ServerSegmentsTableDescriptor implements SystemTableDescriptor
+{
+ public static final String TABLE_NAME = "server_segments";
+ public static final RowSignature ROW_SIGNATURE = RowSignature
+ .builder()
+ .add("server", ColumnType.STRING)
+ .add("segment_id", ColumnType.STRING)
+ .build();
+
+ private static final Set<NodeRole> NODE_ROLES = Set.of(NodeRole.BROKER);
+ private static final SystemTableRowAuthorizer ROW_AUTHORIZER = (rows, authenticationResult, authorizerMapper) -> {
+ final AuthorizationResult authorizationResult = AuthorizationUtils.authorizeAllResourceActions(
+ authenticationResult,
+ Collections.singletonList(new ResourceAction(Resource.STATE_RESOURCE, Action.READ)),
+ authorizerMapper
+ );
+ if (!authorizationResult.allowAccessWithNoRestriction()) {
+ throw new ForbiddenException(
+ "Insufficient permission to view servers: " + authorizationResult.getErrorMessage()
+ );
+ }
+ return rows;
+ };
+
+ @Override
+ public String getTableName()
+ {
+ return TABLE_NAME;
+ }
+
+ @Override
+ public Set<NodeRole> getNodeRoles()
+ {
+ return NODE_ROLES;
+ }
+
+ @Override
+ public SystemTableRoutingMode getRoutingMode()
+ {
+ return SystemTableRoutingMode.LOCAL_ONLY;
+ }
+
+ @Override
+ public RowSignature getRowSignature()
+ {
+ return ROW_SIGNATURE;
+ }
+
+ @Override
+ public SystemTableRowAuthorizer getRowAuthorizer()
+ {
+ return ROW_AUTHORIZER;
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java b/server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java
index cc235f17e52..676e028dcc3 100644
--- a/server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java
+++ b/server/src/main/java/org/apache/druid/server/system/table/SystemTableDataProvider.java
@@ -30,7 +30,7 @@ import java.util.List;
import java.util.Optional;
/**
- * Supplies storage-prefiltered rows authorized for the internal caller of one native system table.
+ * Supplies storage-prefiltered rows authorized for the caller of one native system table.
* The implementation is deployed in related service.
* For example, the data provider of sys.tasks is deployed in overlord module
* */
@@ -56,7 +56,7 @@ public interface SystemTableDataProvider
Iterable<Object[]> getRows(
@NotNull List<DimFilter> filters,
- AuthenticationResult internalAuthenticationResult
+ AuthenticationResult authenticationResult
);
/**
diff --git a/server/src/test/java/org/apache/druid/server/system/ServerSegmentsTableDataProviderTest.java b/server/src/test/java/org/apache/druid/server/system/ServerSegmentsTableDataProviderTest.java
new file mode 100644
index 00000000000..c8fd6df6dcc
--- /dev/null
+++ b/server/src/test/java/org/apache/druid/server/system/ServerSegmentsTableDataProviderTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.druid.server.system;
+
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.client.TimelineServerView;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.query.BatchedInlineDataSource;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.security.Access;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.ForbiddenException;
+import org.apache.druid.server.security.ResourceType;
+import org.apache.druid.server.system.table.ServerSegmentsTableDataProvider;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
+import org.apache.druid.server.system.table.SystemTableQueryRequest;
+import org.apache.druid.server.system.table.SystemTableRoutingMode;
+import org.apache.druid.timeline.DataSegment;
+import org.easymock.EasyMock;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+public class ServerSegmentsTableDataProviderTest
+{
+ private static final AuthenticationResult AUTHENTICATION_RESULT =
+ new AuthenticationResult("test-user", "test", null, null);
+
+ @Test
+ public void testReturnsAuthorizedSegmentsInBatchedDataSource()
+ {
+ final DataSegment allowedSegment = segment("allowed", "2024-01-01/2024-01-02");
+ final DataSegment deniedSegment = segment("denied", "2024-01-02/2024-01-03");
+ final DruidServer server = new DruidServer(
+ "historical",
+ "historical:8083",
+ null,
+ 1_000L,
+ null,
+ ServerType.HISTORICAL,
+ "default",
+ 0
+ );
+ server.addDataSegment(allowedSegment);
+ server.addDataSegment(deniedSegment);
+
+ final TimelineServerView serverView = EasyMock.mock(TimelineServerView.class);
+ EasyMock.expect(serverView.getDruidServers()).andReturn(List.of(server.toImmutableDruidServer())).once();
+ EasyMock.replay(serverView);
+
+ final ServerSegmentsTableDataProvider provider = new ServerSegmentsTableDataProvider(
+ serverView,
+ authorizerMapper(true)
+ );
+ final Iterable<Object[]> authorizedRows = provider.getRows(Collections.emptyList(), AUTHENTICATION_RESULT);
+ final RowSignature projectedSignature = RowSignature.builder()
+ .add("segment_id", ServerSegmentsTableDescriptor.ROW_SIGNATURE
+ .getColumnType("segment_id").orElseThrow())
+ .build();
+ final DataSource dataSource = provider.getAuthorizedDataSource(
+ new SystemTableQueryRequest(List.of("segment_id"), projectedSignature),
+ authorizedRows
+ ).orElseThrow();
+
+ final BatchedInlineDataSource batchedDataSource = Assertions.assertInstanceOf(
+ BatchedInlineDataSource.class,
+ dataSource
+ );
+ final List<Object[]> rows = toRows(batchedDataSource.getRows());
+ Assertions.assertEquals(projectedSignature, batchedDataSource.getRowSignature());
+ Assertions.assertEquals(1, rows.size());
+ Assertions.assertArrayEquals(new Object[]{allowedSegment.getId().toString()}, rows.get(0));
+ EasyMock.verify(serverView);
+ }
+
+ @Test
+ public void testDescriptorRejectsRequestWithoutStateRead()
+ {
+ final ServerSegmentsTableDescriptor descriptor = new ServerSegmentsTableDescriptor();
+
+ Assertions.assertThrows(
+ ForbiddenException.class,
+ () -> descriptor.getRowAuthorizer().filterAuthorizedRows(
+ Collections.emptyList(),
+ AUTHENTICATION_RESULT,
+ authorizerMapper(false)
+ )
+ );
+ }
+
+ @Test
+ public void testDescriptorRunsLocallyOnBroker()
+ {
+ final ServerSegmentsTableDescriptor descriptor = new ServerSegmentsTableDescriptor();
+
+ Assertions.assertEquals(Set.of(NodeRole.BROKER), descriptor.getNodeRoles());
+ Assertions.assertEquals(SystemTableRoutingMode.LOCAL_ONLY, descriptor.getRoutingMode());
+ Assertions.assertEquals(List.of("server", "segment_id"), descriptor.getRowSignature().getColumnNames());
+ }
+
+ private static AuthorizerMapper authorizerMapper(final boolean allowState)
+ {
+ return new AuthorizerMapper(null)
+ {
+ @Override
+ public org.apache.druid.server.security.Authorizer getAuthorizer(final String name)
+ {
+ return (authenticationResult, resource, action) ->
+ (allowState && ResourceType.STATE.equals(resource.getType())) || "allowed".equals(resource.getName())
+ ? Access.OK
+ : Access.DENIED;
+ }
+ };
+ }
+
+ private static DataSegment segment(final String dataSource, final String interval)
+ {
+ return DataSegment.builder()
+ .dataSource(dataSource)
+ .interval(Intervals.of(interval))
+ .version("v1")
+ .size(1L)
+ .build();
+ }
+
+ private static List<Object[]> toRows(final Iterable<Object[]> rows)
+ {
+ final List<Object[]> result = new ArrayList<>();
+ rows.forEach(result::add);
+ return result;
+ }
+}
diff --git a/services/src/main/java/org/apache/druid/cli/CliBroker.java b/services/src/main/java/org/apache/druid/cli/CliBroker.java
index e37b48e5e4c..1e82a99b5f1 100644
--- a/services/src/main/java/org/apache/druid/cli/CliBroker.java
+++ b/services/src/main/java/org/apache/druid/cli/CliBroker.java
@@ -83,6 +83,8 @@ import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
import org.apache.druid.server.metrics.SubqueryCountStatsProvider;
import org.apache.druid.server.router.TieredBrokerConfig;
import org.apache.druid.server.system.table.SegmentsTableDescriptor;
+import org.apache.druid.server.system.table.ServerSegmentsTableDataProvider;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
import org.apache.druid.server.system.table.SystemTableDataProvider;
import org.apache.druid.sql.calcite.schema.MetadataSegmentView;
import org.apache.druid.sql.calcite.schema.SegmentsTableDataProvider;
@@ -134,6 +136,10 @@ public class CliBroker extends ServerRunnable
.addBinding(SegmentsTableDescriptor.TABLE_NAME)
.to(SegmentsTableDataProvider.class)
.in(LazySingleton.class);
+ MapBinder.newMapBinder(binder, String.class, SystemTableDataProvider.class)
+ .addBinding(ServerSegmentsTableDescriptor.TABLE_NAME)
+ .to(ServerSegmentsTableDataProvider.class)
+ .in(LazySingleton.class);
binder.bindConstant().annotatedWith(Names.named("serviceName")).to(
TieredBrokerConfig.DEFAULT_BROKER_SERVICE_NAME
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServerSegmentsTable.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServerSegmentsTable.java
new file mode 100644
index 00000000000..4b3f831d3f7
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/NativeServerSegmentsTable.java
@@ -0,0 +1,72 @@
+/*
+ * 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.druid.sql.calcite.schema;
+
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalTableScan;
+import org.apache.calcite.schema.Schema;
+import org.apache.druid.query.DataSource;
+import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+/** Native-query representation of {@code sys.server_segments}. */
+class NativeServerSegmentsTable extends DruidTable
+{
+ private static final DataSource DATA_SOURCE = new SystemTableDataSource(
+ ServerSegmentsTableDescriptor.TABLE_NAME
+ );
+
+ NativeServerSegmentsTable()
+ {
+ super(ServerSegmentsTableDescriptor.ROW_SIGNATURE);
+ }
+
+ @Override
+ public DataSource getDataSource()
+ {
+ return DATA_SOURCE;
+ }
+
+ @Override
+ public boolean isJoinable()
+ {
+ return false;
+ }
+
+ @Override
+ public boolean isBroadcast()
+ {
+ return false;
+ }
+
+ @Override
+ public Schema.TableType getJdbcTableType()
+ {
+ return Schema.TableType.SYSTEM_TABLE;
+ }
+
+ @Override
+ public RelNode toRel(final RelOptTable.ToRelContext context, final RelOptTable table)
+ {
+ return LogicalTableScan.create(context.getCluster(), table, context.getTableHints());
+ }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
index 52379312bad..b43e0a415dc 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
@@ -70,6 +70,7 @@ import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.server.security.ResourceType;
import org.apache.druid.server.system.table.SegmentsTableDescriptor;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
import org.apache.druid.sql.calcite.planner.PlannerConfig;
import org.apache.druid.sql.calcite.planner.PlannerContext;
import org.apache.druid.sql.calcite.run.NativeSqlEngine;
@@ -98,7 +99,7 @@ public class SystemSchema extends AbstractTableSchema
{
public static final String SEGMENTS_TABLE = SegmentsTableDescriptor.TABLE_NAME;
public static final String SERVERS_TABLE = "servers";
- public static final String SERVER_SEGMENTS_TABLE = "server_segments";
+ public static final String SERVER_SEGMENTS_TABLE = ServerSegmentsTableDescriptor.TABLE_NAME;
public static final String TASKS_TABLE = "tasks";
public static final String SUPERVISOR_TABLE = "supervisors";
public static final String QUERIES_TABLE = "queries";
@@ -151,12 +152,6 @@ public class SystemSchema extends AbstractTableSchema
.add("total_memory", ColumnType.LONG)
.build();
- static final RowSignature SERVER_SEGMENTS_SIGNATURE = RowSignature
- .builder()
- .add("server", ColumnType.STRING)
- .add("segment_id", ColumnType.STRING)
- .build();
-
static final RowSignature TASKS_SIGNATURE = RowSignature
.builder()
.add("task_id", ColumnType.STRING)
@@ -669,7 +664,7 @@ public class SystemSchema extends AbstractTableSchema
/**
* This table contains row per segment per server.
*/
- static class ServerSegmentsTable extends AbstractTable implements ScannableTable
+ static class ServerSegmentsTable extends AbstractTable implements ScannableTable, NativeSystemTable
{
private final TimelineServerView serverView;
private final AuthorizerMapper authorizerMapper;
@@ -689,7 +684,7 @@ public class SystemSchema extends AbstractTableSchema
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory)
{
- return RowSignatures.toRelDataType(SERVER_SEGMENTS_SIGNATURE, typeFactory);
+ return RowSignatures.toRelDataType(ServerSegmentsTableDescriptor.ROW_SIGNATURE, typeFactory);
}
@Override
@@ -698,6 +693,12 @@ public class SystemSchema extends AbstractTableSchema
return TableType.SYSTEM_TABLE;
}
+ @Override
+ public DruidTable asNativeTable()
+ {
+ return new NativeServerSegmentsTable();
+ }
+
@Override
public Enumerable<Object[]> scan(DataContext root)
{
@@ -705,7 +706,7 @@ public class SystemSchema extends AbstractTableSchema
final List<Object[]> rows = new ArrayList<>();
final List<ImmutableDruidServer> druidServers = serverView.getDruidServers();
- final int serverSegmentsTableSize = SERVER_SEGMENTS_SIGNATURE.size();
+ final int serverSegmentsTableSize = ServerSegmentsTableDescriptor.ROW_SIGNATURE.size();
for (ImmutableDruidServer druidServer : druidServers) {
final Iterable<DataSegment> authorizedServerSegments = AuthorizationUtils.filterAuthorizedResources(
authenticationResult,
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
index 0a83956c152..f63d04b71f4 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
@@ -107,6 +107,7 @@ import org.apache.druid.server.security.Authorizer;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.NoopEscalator;
import org.apache.druid.server.security.ResourceType;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
import org.apache.druid.sql.calcite.planner.PlannerConfig;
import org.apache.druid.sql.calcite.run.SqlEngine;
import org.apache.druid.sql.calcite.schema.SystemSchema.QueriesTable;
@@ -1389,7 +1390,7 @@ public class SystemSchemaTest extends CalciteTestBase
Assertions.assertEquals("test5_2015-01-01T00:00:00.000Z_2016-01-01T00:00:00.000Z_version5", row4[1].toString());
// Verify value types.
- verifyTypes(rows, SystemSchema.SERVER_SEGMENTS_SIGNATURE);
+ verifyTypes(rows, ServerSegmentsTableDescriptor.ROW_SIGNATURE);
}
@Test
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java
index 7ce3772ab8e..f7672282a8c 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemTableDataProviderTest.java
@@ -22,6 +22,7 @@ package org.apache.druid.sql.calcite.schema;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.schema.Schema;
import org.apache.druid.query.SystemTableDataSource;
+import org.apache.druid.server.system.table.ServerSegmentsTableDescriptor;
import org.apache.druid.sql.calcite.planner.PlannerContext;
import org.apache.druid.sql.calcite.run.NativeSqlEngine;
import org.apache.druid.sql.calcite.run.SqlEngine;
@@ -42,6 +43,16 @@ public class SystemTableDataProviderTest
Assertions.assertFalse(segments.isBroadcast());
Assertions.assertEquals(Schema.TableType.SYSTEM_TABLE, segments.getJdbcTableType());
+ final NativeServerSegmentsTable serverSegments = new NativeServerSegmentsTable();
+ Assertions.assertEquals(
+ "server_segments",
+ ((SystemTableDataSource) serverSegments.getDataSource()).getTable()
+ );
+ Assertions.assertEquals(ServerSegmentsTableDescriptor.ROW_SIGNATURE, serverSegments.getRowSignature());
+ Assertions.assertFalse(serverSegments.isJoinable());
+ Assertions.assertFalse(serverSegments.isBroadcast());
+ Assertions.assertEquals(Schema.TableType.SYSTEM_TABLE, serverSegments.getJdbcTableType());
+
final NativeServerPropertiesTable serverProperties = new NativeServerPropertiesTable();
Assertions.assertEquals(
"server_properties",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org