Introduce SystemSchema tables (#5989)

benchmarks/src/main/java/org/apache/druid/benchmark/query/SqlBenchmark.java

@@ -44,6 +44,8 @@
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
+import org.apache.druid.sql.calcite.schema.DruidSchema;
+import org.apache.druid.sql.calcite.schema.SystemSchema;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.SpecificSegmentsQuerySegmentWalker;
 import org.apache.druid.timeline.DataSegment;
@@ -111,10 +113,12 @@ public void setup()
         .createQueryRunnerFactoryConglomerate();
     final QueryRunnerFactoryConglomerate conglomerate = conglomerateCloserPair.lhs;
     final PlannerConfig plannerConfig = new PlannerConfig();
-
+    final DruidSchema druidSchema = CalciteTests.createMockSchema(conglomerate, walker, plannerConfig);
+    final SystemSchema systemSchema = CalciteTests.createMockSystemSchema(druidSchema, walker);
     this.walker = new SpecificSegmentsQuerySegmentWalker(conglomerate).add(dataSegment, index);
     plannerFactory = new PlannerFactory(
-        CalciteTests.createMockSchema(conglomerate, walker, plannerConfig),
+        druidSchema,
+        systemSchema,
         CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),

docs/content/querying/sql.md

@@ -477,6 +477,11 @@ 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 Sys Schema.
+Information schema provides details about table and column types. The "sys" schema provides information about Druid internals like segments/tasks/servers.
+
+## 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:
@@ -528,6 +533,101 @@ 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
+
+The "sys" schema provides visibility into 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 table provides details on all Druid segments, whether they are published yet or not.
+
+
+|Column|Notes|
+|------|-----|
+|segment_id|Unique segment identifier|
+|datasource|Name of datasource|
+|start|Interval start time (in ISO 8601 format)|
+|end|Interval end time (in ISO 8601 format)|
+|size|Size of segment in bytes|
+|version|Version string (generally an ISO8601 timestamp corresponding to when the segment set was first started). Higher version means the more recently created segment. Version comparing is based on string comparison.|
+|partition_num|Partition number (an integer, unique within a datasource+interval+version; may not necessarily be contiguous)|
+|num_replicas|Number of replicas of this segment currently being served|
+|num_rows|Number of rows in current segment, this value could be null if unkown to broker at query time|
+|is_published|Boolean is represented as long type where 1 = true, 0 = false. 1 represents this segment has been published to the metadata store|
+|is_available|Boolean is represented as long type where 1 = true, 0 = false. 1 if this segment is currently being served by any server(historical or realtime)|
+|is_realtime|Boolean is represented as long type where 1 = true, 0 = false. 1 if this segment is being served on any type of realtime tasks|
+|payload|JSON-serialized data segment payload|
+
+### SERVERS table
+Servers table lists all data servers(any server that hosts a segment). It includes both historicals and peons.
+
+|Column|Notes|
+|------|-----|
+|server|Server name in the form host:port|
+|host|Hostname of the server|
+|plaintext_port|Unsecured port of the server, or -1 if plaintext traffic is disabled|
+|tls_port|TLS port of the server, or -1 if TLS is disabled|
+|server_type|Type of Druid service. Possible values include: historical, realtime and indexer_executor(peon).|
+|tier|Distribution tier see [druid.server.tier](#../configuration/index.html#Historical-General-Configuration)|
+|current_size|Current size of segments in bytes on this server|
+|max_size|Max size in bytes this server recommends to assign to segments see [druid.server.maxSize](#../configuration/index.html#Historical-General-Configuration)|
+
+To retrieve information about all servers, use the query:
+```sql
+SELECT * FROM sys.servers;
+```
+
+### SERVER_SEGMENTS table
+
+SERVER_SEGMENTS is used to join servers with segments table
+
+|Column|Notes|
+|------|-----|
+|server|Server name in format host:port (Primary key of [servers table](#SERVERS-table))|
+|segment_id|Segment identifier (Primary key of [segments table](#SEGMENTS-table))|
+
+JOIN between "servers" and "segments" can be used to query the number of segments for a specific datasource, 
+grouped by server, example query:
+```sql
+SELECT count(segments.segment_id) as num_segments from sys.segments as segments 
+INNER JOIN sys.server_segments as server_segments 
+ON segments.segment_id  = server_segments.segment_id 
+INNER JOIN sys.servers as servers 
+ON servers.server = server_segments.server
+WHERE segments.datasource = 'wikipedia' 
+GROUP BY servers.server;
+```
+
+### TASKS table
+
+The tasks table provides information about active and recently-completed indexing tasks. For more information 
+check out [ingestion tasks](#../ingestion/tasks.html)
+
+|Column|Notes|
+|------|-----|
+|task_id|Unique task identifier|
+|type|Task type, for example this value is "index" for indexing tasks. See [tasks-overview](../ingestion/tasks.md)|
+|datasource|Datasource name being indexed|
+|created_time|Timestamp in ISO8601 format corresponding to when the ingestion task was created. Note that this value is populated for completed and waiting tasks. For running and pending tasks this value is set to 1970-01-01T00:00:00Z|
+|queue_insertion_time|Timestamp in ISO8601 format corresponding to when this task was added to the queue on the overlord|
+|status|Status of a task can be RUNNING, FAILED, SUCCESS|
+|runner_status|Runner status of a completed task would be NONE, for in-progress tasks this can be RUNNING, WAITING, PENDING|
+|duration|Time it took to finish the task in milliseconds, this value is present only for completed tasks|
+|location|Server name where this task is running in the format host:port, this information is present only for RUNNING tasks|
+|host|Hostname of the server where task is running|
+|plaintext_port|Unsecured port of the server, or -1 if plaintext traffic is disabled|
+|tls_port|TLS port of the server, or -1 if TLS is disabled|
+|error_msg|Detailed error message in case of FAILED tasks|
+
+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/org/apache/druid/query/aggregation/histogram/sql/QuantileSqlAggregatorTest.java

@@ -60,6 +60,7 @@
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.planner.PlannerResult;
 import org.apache.druid.sql.calcite.schema.DruidSchema;
+import org.apache.druid.sql.calcite.schema.SystemSchema;
 import org.apache.druid.sql.calcite.util.CalciteTestBase;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.QueryLogHook;
@@ -150,13 +151,15 @@ public void setUp() throws Exception
 
     final PlannerConfig plannerConfig = new PlannerConfig();
     final DruidSchema druidSchema = CalciteTests.createMockSchema(conglomerate, walker, plannerConfig);
+    final SystemSchema systemSchema = CalciteTests.createMockSystemSchema(druidSchema, walker);
     final DruidOperatorTable operatorTable = new DruidOperatorTable(
         ImmutableSet.of(new QuantileSqlAggregator()),
         ImmutableSet.of()
     );
 
     plannerFactory = new PlannerFactory(
         druidSchema,
+        systemSchema,
         CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
         operatorTable,
         CalciteTests.createExprMacroTable(),

server/src/main/java/org/apache/druid/client/BrokerServerView.java

@@ -45,12 +45,14 @@
 import org.apache.druid.timeline.partition.PartitionChunk;
 
 import javax.annotation.Nullable;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  */
@@ -322,4 +324,12 @@ private void runTimelineCallbacks(final Function<TimelineCallback, CallbackActio
       );
     }
   }
+
+  @Override
+  public List<ImmutableDruidServer> getDruidServers()
+  {
+    return clients.values().stream()
+                  .map(queryableDruidServer -> queryableDruidServer.getServer().toImmutableDruidServer())
+                  .collect(Collectors.toList());
+  }
 }

server/src/main/java/org/apache/druid/client/DirectDruidClient.java

@@ -19,9 +19,6 @@
 
 package org.apache.druid.client;
 
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonToken;
-import com.fasterxml.jackson.core.ObjectCodec;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.type.TypeFactory;
@@ -32,7 +29,6 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.RE;
 import org.apache.druid.java.util.common.StringUtils;
@@ -52,14 +48,12 @@
 import org.apache.druid.query.BySegmentResultValueClass;
 import org.apache.druid.query.Query;
 import org.apache.druid.query.QueryContexts;
-import org.apache.druid.query.QueryInterruptedException;
 import org.apache.druid.query.QueryMetrics;
 import org.apache.druid.query.QueryPlus;
 import org.apache.druid.query.QueryRunner;
 import org.apache.druid.query.QueryToolChest;
 import org.apache.druid.query.QueryToolChestWarehouse;
 import org.apache.druid.query.QueryWatcher;
-import org.apache.druid.query.ResourceLimitExceededException;
 import org.apache.druid.query.Result;
 import org.apache.druid.query.aggregation.MetricManipulatorFns;
 import org.jboss.netty.buffer.ChannelBuffer;
@@ -71,20 +65,16 @@
 import org.joda.time.Duration;
 
 import javax.ws.rs.core.MediaType;
-import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.SequenceInputStream;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.Enumeration;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -550,7 +540,7 @@ public void onFailure(Throwable t)
           @Override
           public JsonParserIterator<T> make()
           {
-            return new JsonParserIterator<T>(typeRef, future, url, query);
+            return new JsonParserIterator<T>(typeRef, future, url, query, host, objectMapper);
           }
 
           @Override
@@ -576,113 +566,6 @@ public void cleanup(JsonParserIterator<T> iterFromMake)
     return retVal;
   }
 
-  private class JsonParserIterator<T> implements Iterator<T>, Closeable
-  {
-    private JsonParser jp;
-    private ObjectCodec objectCodec;
-    private final JavaType typeRef;
-    private final Future<InputStream> future;
-    private final Query<T> query;
-    private final String url;
-
-    public JsonParserIterator(JavaType typeRef, Future<InputStream> future, String url, Query<T> query)
-    {
-      this.typeRef = typeRef;
-      this.future = future;
-      this.url = url;
-      this.query = query;
-      jp = null;
-    }
-
-    @Override
-    public boolean hasNext()
-    {
-      init();
-
-      if (jp.isClosed()) {
-        return false;
-      }
-      if (jp.getCurrentToken() == JsonToken.END_ARRAY) {
-        CloseQuietly.close(jp);
-        return false;
-      }
-
-      return true;
-    }
-
-    @Override
-    public T next()
-    {
-      init();
-
-      try {
-        final T retVal = objectCodec.readValue(jp, typeRef);
-        jp.nextToken();
-        return retVal;
-      }
-      catch (IOException e) {
-        throw Throwables.propagate(e);
-      }
-    }
-
-    @Override
-    public void remove()
-    {
-      throw new UnsupportedOperationException();
-    }
-
-    private void init()
-    {
-      if (jp == null) {
-        try {
-          InputStream is = future.get();
-          if (is == null) {
-            throw new QueryInterruptedException(
-                new ResourceLimitExceededException(
-                    "query[%s] url[%s] timed out or max bytes limit reached.",
-                    query.getId(),
-                    url
-                ),
-                host
-            );
-          } else {
-            jp = objectMapper.getFactory().createParser(is);
-          }
-          final JsonToken nextToken = jp.nextToken();
-          if (nextToken == JsonToken.START_OBJECT) {
-            QueryInterruptedException cause = jp.getCodec().readValue(jp, QueryInterruptedException.class);
-            throw new QueryInterruptedException(cause, host);
-          } else if (nextToken != JsonToken.START_ARRAY) {
-            throw new IAE("Next token wasn't a START_ARRAY, was[%s] from url [%s]", jp.getCurrentToken(), url);
-          } else {
-            jp.nextToken();
-            objectCodec = jp.getCodec();
-          }
-        }
-        catch (IOException | InterruptedException | ExecutionException e) {
-          throw new RE(
-              e,
-              "Failure getting results for query[%s] url[%s] because of [%s]",
-              query.getId(),
-              url,
-              e.getMessage()
-          );
-        }
-        catch (CancellationException e) {
-          throw new QueryInterruptedException(e, host);
-        }
-      }
-    }
-
-    @Override
-    public void close() throws IOException
-    {
-      if (jp != null) {
-        jp.close();
-      }
-    }
-  }
-
   @Override
   public String toString()
   {

server/src/main/java/org/apache/druid/client/ImmutableDruidServer.java

@@ -65,6 +65,16 @@ public String getHost()
     return metadata.getHost();
   }
 
+  public String getHostAndPort()
+  {
+    return metadata.getHostAndPort();
+  }
+
+  public String getHostAndTlsPort()
+  {
+    return metadata.getHostAndTlsPort();
+  }
+
   public long getCurrSize()
   {
     return currSize;