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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
