Introduce SystemSchema tables (#5989)

benchmarks/src/main/java/io/druid/benchmark/query/SqlBenchmark.java

@@ -25,6 +25,7 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.benchmark.datagen.SegmentGenerator;
 import io.druid.data.input.Row;
+import io.druid.discovery.DruidLeaderClient;
 import io.druid.java.util.common.Intervals;
 import io.druid.java.util.common.granularity.Granularities;
 import io.druid.java.util.common.guava.Sequence;
@@ -43,9 +44,11 @@
 import io.druid.sql.calcite.planner.PlannerResult;
 import io.druid.sql.calcite.util.CalciteTests;
 import io.druid.sql.calcite.util.SpecificSegmentsQuerySegmentWalker;
+import io.druid.sql.calcite.util.TestServerInventoryView;
 import io.druid.timeline.DataSegment;
 import io.druid.timeline.partition.LinearShardSpec;
 import org.apache.commons.io.FileUtils;
+import org.easymock.EasyMock;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -106,16 +109,20 @@ public void setup()
     final QueryableIndex index = segmentGenerator.generate(dataSegment, schemaInfo, Granularities.NONE, rowsPerSegment);
     final QueryRunnerFactoryConglomerate conglomerate = CalciteTests.queryRunnerFactoryConglomerate();
     final PlannerConfig plannerConfig = new PlannerConfig();
+    final DruidLeaderClient druidLeaderClient = EasyMock.createMock(DruidLeaderClient.class);
 
     this.walker = new SpecificSegmentsQuerySegmentWalker(conglomerate).add(dataSegment, index);
     plannerFactory = new PlannerFactory(
         CalciteTests.createMockSchema(walker, plannerConfig),
+        new TestServerInventoryView(walker.getSegments()),
         CalciteTests.createMockQueryLifecycleFactory(walker),
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,
         AuthTestUtils.TEST_AUTHORIZER_MAPPER,
-        CalciteTests.getJsonMapper()
+        CalciteTests.getJsonMapper(),
+        druidLeaderClient,
+        druidLeaderClient
     );
     groupByQuery = GroupByQuery
         .builder()

docs/content/querying/sql.md

@@ -430,6 +430,10 @@ plan SQL queries. This metadata is cached on broker startup and also updated per
 [SegmentMetadata queries](segmentmetadataquery.html). Background metadata refreshing is triggered by
 segments entering and exiting the cluster, and can also be throttled through configuration.
 
+Druid exposes system information through special system tables. There are two such schemas available : Information Schema and System Schema
+
+## INFORMATION SCHEMA
+
 You can access table and column metadata through JDBC using `connection.getMetaData()`, or through the
 INFORMATION_SCHEMA tables described below. For example, to retrieve metadata for the Druid
 datasource "foo", use the query:
@@ -481,6 +485,77 @@ SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = 'druid' AND TABLE_
 |COLLATION_NAME||
 |JDBC_TYPE|Type code from java.sql.Types (Druid extension)|
 
+## SYSTEM SCHEMA
+
+SYSTEM_TABLES provide visibility into the druid segments, servers and tasks.
+For example to retrieve all segments for datasource "wikipedia", use the query:
+```sql
+select * from SYS.SEGMENTS where DATASOURCE='wikipedia';
+```
+
+### SEGMENTS table
+Segments tables provides details on all the segments, both published and served(but not published).
+
+
+|Column|Notes|
+|------|-----|
+|SEGMENT_ID||
+|DATASOURCE||
+|START||
+|END||
+|IS_PUBLISHED|segment in metadata store|
+|IS_AVAILABLE|segment is being served|
+|IS_REALTIME|segment served on a realtime server|
+|PAYLOAD|jsonified datasegment payload|
+
+### SERVERS table
+
+
+|Column|Notes|
+|------|-----|
+|SERVER||
+|SERVER_TYPE||
+|TIER||
+|CURRENT_SIZE||
+|MAX_SIZE||
+
+To retrieve all servers information, use the query
+```sql
+select * from SYS.SERVERS;
+```
+
+### SEGMENTSERVERS table
+
+SEGMENTSERVERS is used to join SEGMENTS with SERVERS table
+
+|Column|Notes|
+|------|-----|
+|SERVER||
+|SEGMENT_ID||
+
+### TASKS table
+
+TASKS table provides tasks info from overlord.
+
+|Column|Notes|
+|------|-----|
+|TASK_ID||
+|TYPE||
+|DATASOURCE||
+|CREATED_TIME||
+|QUEUE_INSERTION_TIME||
+|STATUS||
+|RUNNER_STATUS||
+|DURATION||
+|LOCATION||
+|ERROR_MSG||
+
+For example, to retrieve tasks information filtered by status, use the query
+```sql
+select * from SYS.TASKS where STATUS='FAILED';
+```
+
+
 ## Server configuration
 
 The Druid SQL server is configured through the following properties on the broker.

extensions-core/histogram/src/test/java/io/druid/query/aggregation/histogram/sql/QuantileSqlAggregatorTest.java

@@ -23,7 +23,9 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
+import io.druid.client.TimelineServerView;
 import io.druid.common.config.NullHandling;
+import io.druid.discovery.DruidLeaderClient;
 import io.druid.java.util.common.granularity.Granularities;
 import io.druid.query.Druids;
 import io.druid.query.QueryDataSource;
@@ -61,8 +63,10 @@
 import io.druid.sql.calcite.util.CalciteTests;
 import io.druid.sql.calcite.util.QueryLogHook;
 import io.druid.sql.calcite.util.SpecificSegmentsQuerySegmentWalker;
+import io.druid.sql.calcite.util.TestServerInventoryView;
 import io.druid.timeline.DataSegment;
 import io.druid.timeline.partition.LinearShardSpec;
+import org.easymock.EasyMock;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -126,19 +130,24 @@ public void setUp() throws Exception
 
     final PlannerConfig plannerConfig = new PlannerConfig();
     final DruidSchema druidSchema = CalciteTests.createMockSchema(walker, plannerConfig);
+    final TimelineServerView serverView = new TestServerInventoryView(walker.getSegments());
+    final DruidLeaderClient druidLeaderClient = EasyMock.createMock(DruidLeaderClient.class);
     final DruidOperatorTable operatorTable = new DruidOperatorTable(
         ImmutableSet.of(new QuantileSqlAggregator()),
         ImmutableSet.of()
     );
 
     plannerFactory = new PlannerFactory(
         druidSchema,
+        serverView,
         CalciteTests.createMockQueryLifecycleFactory(walker),
         operatorTable,
         CalciteTests.createExprMacroTable(),
         plannerConfig,
         AuthTestUtils.TEST_AUTHORIZER_MAPPER,
-        CalciteTests.getJsonMapper()
+        CalciteTests.getJsonMapper(),
+        druidLeaderClient,
+        druidLeaderClient
     );
   }
 

server/src/main/java/io/druid/client/BrokerServerView.java

@@ -322,4 +322,10 @@ private void runTimelineCallbacks(final Function<TimelineCallback, CallbackActio
       );
     }
   }
+
+  @Override
+  public Map<String, QueryableDruidServer> getClients()
+  {
+    return clients;
+  }
 }

server/src/main/java/io/druid/client/TimelineServerView.java

@@ -19,6 +19,7 @@
 
 package io.druid.client;
 
+import io.druid.client.selector.QueryableDruidServer;
 import io.druid.client.selector.ServerSelector;
 import io.druid.query.DataSource;
 import io.druid.query.QueryRunner;
@@ -27,6 +28,7 @@
 import io.druid.timeline.TimelineLookup;
 
 import javax.annotation.Nullable;
+import java.util.Map;
 import java.util.concurrent.Executor;
 
 /**
@@ -36,6 +38,9 @@ public interface TimelineServerView extends ServerView
   @Nullable
   TimelineLookup<String, ServerSelector> getTimeline(DataSource dataSource);
 
+  @Nullable
+  Map<String, QueryableDruidServer> getClients();
+
   <T> QueryRunner<T> getQueryRunner(DruidServer server);
 
   /**

server/src/test/java/io/druid/client/CachingClusteredClientFunctionalityTest.java

@@ -49,6 +49,7 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import javax.annotation.Nullable;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -225,6 +226,13 @@ public VersionedIntervalTimeline<String, ServerSelector> getTimeline(DataSource
             return timeline;
           }
 
+          @Nullable
+          @Override
+          public Map<String, QueryableDruidServer> getClients()
+          {
+            throw new UnsupportedOperationException();
+          }
+
           @Override
           public void registerTimelineCallback(final Executor exec, final TimelineCallback callback)
           {

server/src/test/java/io/druid/client/CachingClusteredClientTest.java

@@ -2656,6 +2656,13 @@ public VersionedIntervalTimeline<String, ServerSelector> getTimeline(DataSource
             return timeline;
           }
 
+          @Nullable
+          @Override
+          public Map<String, QueryableDruidServer> getClients()
+          {
+            throw new UnsupportedOperationException();
+          }
+
           @Override
           public <T> QueryRunner<T> getQueryRunner(DruidServer server)
           {

sql/src/main/java/io/druid/sql/calcite/planner/Calcites.java

@@ -19,9 +19,13 @@
 
 package io.druid.sql.calcite.planner;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
 import com.google.common.io.BaseEncoding;
 import com.google.common.primitives.Chars;
+import io.druid.client.BrokerServerView;
+import io.druid.client.TimelineServerView;
+import io.druid.discovery.DruidLeaderClient;
 import io.druid.java.util.common.DateTimes;
 import io.druid.java.util.common.IAE;
 import io.druid.java.util.common.ISE;
@@ -32,6 +36,7 @@
 import io.druid.server.security.AuthorizerMapper;
 import io.druid.sql.calcite.schema.DruidSchema;
 import io.druid.sql.calcite.schema.InformationSchema;
+import io.druid.sql.calcite.schema.SystemSchema;
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
@@ -98,11 +103,30 @@ public static Charset defaultCharset()
     return DEFAULT_CHARSET;
   }
 
-  public static SchemaPlus createRootSchema(final Schema druidSchema, final AuthorizerMapper authorizerMapper)
+  public static SchemaPlus createRootSchema(
+      final TimelineServerView serverView,
+      final Schema druidSchema,
+      final AuthorizerMapper authorizerMapper,
+      final DruidLeaderClient coordinatorDruidLeaderClient,
+      final DruidLeaderClient overlordDruidLeaderClient,
+      final ObjectMapper jsonMapper
+  )
   {
     final SchemaPlus rootSchema = CalciteSchema.createRootSchema(false, false).plus();
     rootSchema.add(DruidSchema.NAME, druidSchema);
     rootSchema.add(InformationSchema.NAME, new InformationSchema(rootSchema, authorizerMapper));
+    if (serverView instanceof BrokerServerView) {
+      rootSchema.add(
+          SystemSchema.NAME,
+          new SystemSchema(
+              (BrokerServerView) serverView,
+              authorizerMapper,
+              coordinatorDruidLeaderClient,
+              overlordDruidLeaderClient,
+              jsonMapper
+          )
+      );
+    }
     return rootSchema;
   }
 

sql/src/main/java/io/druid/sql/calcite/planner/PlannerFactory.java

@@ -21,6 +21,10 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.inject.Inject;
+import io.druid.client.TimelineServerView;
+import io.druid.client.coordinator.Coordinator;
+import io.druid.client.indexing.IndexingService;
+import io.druid.discovery.DruidLeaderClient;
 import io.druid.guice.annotations.Json;
 import io.druid.math.expr.ExprMacroTable;
 import io.druid.server.QueryLifecycleFactory;
@@ -57,36 +61,52 @@ public class PlannerFactory
       .build();
 
   private final DruidSchema druidSchema;
+  private final TimelineServerView serverView;
   private final QueryLifecycleFactory queryLifecycleFactory;
   private final DruidOperatorTable operatorTable;
   private final ExprMacroTable macroTable;
   private final PlannerConfig plannerConfig;
   private final ObjectMapper jsonMapper;
   private final AuthorizerMapper authorizerMapper;
+  private final DruidLeaderClient coordinatorDruidLeaderClient;
+  private final DruidLeaderClient overlordDruidLeaderClient;
 
   @Inject
   public PlannerFactory(
       final DruidSchema druidSchema,
+      final TimelineServerView serverView,
       final QueryLifecycleFactory queryLifecycleFactory,
       final DruidOperatorTable operatorTable,
       final ExprMacroTable macroTable,
       final PlannerConfig plannerConfig,
       final AuthorizerMapper authorizerMapper,
-      final @Json ObjectMapper jsonMapper
+      final @Json ObjectMapper jsonMapper,
+      final @Coordinator DruidLeaderClient coordinatorDruidLeaderClient,
+      final @IndexingService DruidLeaderClient overlordDruidLeaderClient
   )
   {
     this.druidSchema = druidSchema;
+    this.serverView = serverView;
     this.queryLifecycleFactory = queryLifecycleFactory;
     this.operatorTable = operatorTable;
     this.macroTable = macroTable;
     this.plannerConfig = plannerConfig;
     this.authorizerMapper = authorizerMapper;
     this.jsonMapper = jsonMapper;
+    this.coordinatorDruidLeaderClient = coordinatorDruidLeaderClient;
+    this.overlordDruidLeaderClient = overlordDruidLeaderClient;
   }
 
   public DruidPlanner createPlanner(final Map<String, Object> queryContext)
   {
-    final SchemaPlus rootSchema = Calcites.createRootSchema(druidSchema, authorizerMapper);
+    final SchemaPlus rootSchema = Calcites.createRootSchema(
+        serverView,
+        druidSchema,
+        authorizerMapper,
+        coordinatorDruidLeaderClient,
+        overlordDruidLeaderClient,
+        jsonMapper
+    );
     final PlannerContext plannerContext = PlannerContext.create(
         operatorTable,
         macroTable,

sql/src/main/java/io/druid/sql/calcite/schema/DruidSchema.java

@@ -91,7 +91,6 @@ public class DruidSchema extends AbstractSchema
   private static final int MAX_SEGMENTS_PER_QUERY = 15000;
 
   private final QueryLifecycleFactory queryLifecycleFactory;
-  private final TimelineServerView serverView;
   private final PlannerConfig config;
   private final ViewManager viewManager;
   private final ExecutorService cacheExec;
@@ -134,7 +133,7 @@ public DruidSchema(
   )
   {
     this.queryLifecycleFactory = Preconditions.checkNotNull(queryLifecycleFactory, "queryLifecycleFactory");
-    this.serverView = Preconditions.checkNotNull(serverView, "serverView");
+    Preconditions.checkNotNull(serverView, "serverView");
     this.config = Preconditions.checkNotNull(config, "config");
     this.viewManager = Preconditions.checkNotNull(viewManager, "viewManager");
     this.cacheExec = ScheduledExecutors.fixed(1, "DruidSchema-Cache-%d");