projections + compaction + catalog + msq

benchmarks/src/test/java/org/apache/druid/server/coordinator/NewestSegmentFirstPolicyBenchmark.java

@@ -93,7 +93,7 @@ public void setup()
       final String dataSource = DATA_SOURCE_PREFIX + i;
       compactionConfigs.put(
           dataSource,
-          DataSourceCompactionConfig
+          InlineSchemaDataSourceCompactionConfig
               .builder()
               .forDataSource(dataSource)
               .withTaskPriority(0)

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/guice/CatalogBrokerModule.java → extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/guice/CatalogClientModule.java

@@ -20,55 +20,59 @@
 package org.apache.druid.catalog.guice;
 
 import com.google.inject.Binder;
+import org.apache.druid.catalog.MetadataCatalog;
 import org.apache.druid.catalog.http.CatalogListenerResource;
 import org.apache.druid.catalog.model.SchemaRegistry;
 import org.apache.druid.catalog.model.SchemaRegistryImpl;
 import org.apache.druid.catalog.sql.LiveCatalogResolver;
 import org.apache.druid.catalog.sync.CachedMetadataCatalog;
 import org.apache.druid.catalog.sync.CatalogClient;
+import org.apache.druid.catalog.sync.CatalogClientConfig;
+import org.apache.druid.catalog.sync.CatalogSource;
 import org.apache.druid.catalog.sync.CatalogUpdateListener;
 import org.apache.druid.catalog.sync.CatalogUpdateReceiver;
-import org.apache.druid.catalog.sync.MetadataCatalog;
-import org.apache.druid.catalog.sync.MetadataCatalog.CatalogSource;
 import org.apache.druid.discovery.NodeRole;
 import org.apache.druid.guice.Jerseys;
+import org.apache.druid.guice.JsonConfigProvider;
 import org.apache.druid.guice.LazySingleton;
 import org.apache.druid.guice.LifecycleModule;
 import org.apache.druid.guice.ManageLifecycle;
+import org.apache.druid.guice.annotations.ExcludeScope;
 import org.apache.druid.guice.annotations.LoadScope;
 import org.apache.druid.initialization.DruidModule;
 import org.apache.druid.sql.calcite.planner.CatalogResolver;
 
 /**
- * Configures the metadata catalog on the Broker to use a cache
+ * Configures the metadata catalog on the Broker and Overlord to use a cache
  * and network communications for pull and push updates.
  */
-@LoadScope(roles = NodeRole.BROKER_JSON_NAME)
-public class CatalogBrokerModule implements DruidModule
+@LoadScope(roles = {NodeRole.BROKER_JSON_NAME, NodeRole.OVERLORD_JSON_NAME})
+@ExcludeScope(roles = {NodeRole.COORDINATOR_JSON_NAME})
+public class CatalogClientModule implements DruidModule
 {
   @Override
   public void configure(Binder binder)
   {
-    // The Broker (catalog client) uses a cached metadata catalog.
+    // config for catalog client
+    JsonConfigProvider.bind(binder, "druid.catalog.client", CatalogClientConfig.class);
+
+    // The catalog client maintains a cached metadata catalog.
     binder
         .bind(CachedMetadataCatalog.class)
         .in(LazySingleton.class);
-
-    // Broker code accesses he catalog through the
-    // MetadataCatalog interface.
+    // Catalog metadata for the various purposes. This will override the base binding.
     binder
         .bind(MetadataCatalog.class)
         .to(CachedMetadataCatalog.class)
         .in(LazySingleton.class);
 
-    // The cached metadata catalog needs a "pull" source,
-    // which is the network client.
+    // The cached metadata catalog needs a "pull" source, which is the network client.
     binder
         .bind(CatalogSource.class)
         .to(CatalogClient.class)
         .in(LazySingleton.class);
 
-    // The cached metadata catalog is the listener for"push" events.
+    // The cached metadata catalog is the listener for "push" events.
     binder
         .bind(CatalogUpdateListener.class)
         .to(CachedMetadataCatalog.class)
@@ -80,21 +84,19 @@ public void configure(Binder binder)
         .to(SchemaRegistryImpl.class)
         .in(LazySingleton.class);
 
-    // Lifecycle-managed class to prime the metadata cache
+    // Lifecycle-managed class to periodically the metadata cache
     binder
         .bind(CatalogUpdateReceiver.class)
         .in(ManageLifecycle.class);
     LifecycleModule.register(binder, CatalogUpdateReceiver.class);
 
-    // Catalog resolver for the planner. This will override the
-    // base binding.
+    // Catalog resolver for the planner. This will override the base binding.
     binder
         .bind(CatalogResolver.class)
         .to(LiveCatalogResolver.class)
         .in(LazySingleton.class);
 
-    // The listener resource sends to the catalog
-    // listener (the cached catalog.)
+    // The listener resource sends to the catalog listener (the cached catalog.)
     Jerseys.addResource(binder, CatalogListenerResource.class);
   }
 }

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/guice/CatalogCoordinatorModule.java

@@ -21,14 +21,17 @@
 
 import com.fasterxml.jackson.databind.Module;
 import com.google.inject.Binder;
+import org.apache.druid.catalog.MetadataCatalog;
 import org.apache.druid.catalog.http.CatalogResource;
 import org.apache.druid.catalog.model.SchemaRegistry;
 import org.apache.druid.catalog.model.SchemaRegistryImpl;
 import org.apache.druid.catalog.storage.CatalogStorage;
 import org.apache.druid.catalog.storage.MetadataStorageManager;
 import org.apache.druid.catalog.storage.sql.CatalogManager;
 import org.apache.druid.catalog.storage.sql.SQLCatalogManager;
+import org.apache.druid.catalog.sync.CatalogSource;
 import org.apache.druid.catalog.sync.CatalogUpdateNotifier;
+import org.apache.druid.catalog.sync.LocalMetadataCatalog;
 import org.apache.druid.discovery.NodeRole;
 import org.apache.druid.guice.Jerseys;
 import org.apache.druid.guice.LazySingleton;
@@ -66,6 +69,19 @@ public void configure(Binder binder)
         .bind(MetadataStorageManager.class)
         .in(LazySingleton.class);
 
+    binder
+        .bind(CatalogSource.class)
+        .to(CatalogStorage.class)
+        .in(LazySingleton.class);
+
+    binder
+        .bind(LocalMetadataCatalog.class)
+        .in(LazySingleton.class);
+    binder
+        .bind(MetadataCatalog.class)
+        .to(LocalMetadataCatalog.class)
+        .in(LazySingleton.class);
+
     // At present, the set of schemas is fixed.
     binder
         .bind(SchemaRegistry.class)

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java

@@ -20,13 +20,13 @@
 package org.apache.druid.catalog.sql;
 
 import com.google.common.collect.ImmutableSet;
+import org.apache.druid.catalog.MetadataCatalog;
 import org.apache.druid.catalog.model.ColumnSpec;
 import org.apache.druid.catalog.model.Columns;
 import org.apache.druid.catalog.model.ResolvedTable;
 import org.apache.druid.catalog.model.TableId;
 import org.apache.druid.catalog.model.facade.DatasourceFacade;
 import org.apache.druid.catalog.model.table.DatasourceDefn;
-import org.apache.druid.catalog.sync.MetadataCatalog;
 import org.apache.druid.query.TableDataSource;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
@@ -39,7 +39,6 @@
 
 import javax.annotation.Nullable;
 import javax.inject.Inject;
-
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -201,6 +200,12 @@ public boolean ingestRequiresExistingTable()
     return false;
   }
 
+  @Override
+  public MetadataCatalog getMetadataCatalog()
+  {
+    return catalog;
+  }
+
   @Override
   public Set<String> getTableNames(Set<String> datasourceNames)
   {

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/storage/CatalogStorage.java

@@ -29,14 +29,13 @@
 import org.apache.druid.catalog.model.TableId;
 import org.apache.druid.catalog.model.TableMetadata;
 import org.apache.druid.catalog.storage.sql.CatalogManager;
+import org.apache.druid.catalog.sync.CatalogSource;
 import org.apache.druid.catalog.sync.CatalogUpdateListener;
-import org.apache.druid.catalog.sync.MetadataCatalog.CatalogSource;
-import org.apache.druid.catalog.sync.MetadataCatalog.CatalogUpdateProvider;
+import org.apache.druid.catalog.sync.CatalogUpdateProvider;
 import org.apache.druid.guice.annotations.Json;
 
 import javax.annotation.Nullable;
 import javax.inject.Inject;
-
 import java.util.List;
 
 /**

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CachedMetadataCatalog.java

@@ -20,6 +20,7 @@
 package org.apache.druid.catalog.sync;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.catalog.MetadataCatalog;
 import org.apache.druid.catalog.model.ResolvedTable;
 import org.apache.druid.catalog.model.SchemaRegistry;
 import org.apache.druid.catalog.model.SchemaRegistry.SchemaSpec;
@@ -30,7 +31,6 @@
 import org.apache.druid.java.util.common.logger.Logger;
 
 import javax.inject.Inject;
-
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -339,6 +339,8 @@ public void updated(UpdateEvent event)
     SchemaEntry schemaEntry = entryFor(event.table.id().schema());
     if (schemaEntry != null) {
       schemaEntry.update(event);
+    } else {
+      resync();
     }
   }
 
@@ -379,7 +381,7 @@ private SchemaEntry entryFor(String schemaName)
    * Discard any existing cached tables and reload directly from the
    * catalog source. Manages the two schemas which the catalog manages.
    * If the catalog were to manage others, add those here as well.
-   * Done both at Broker startup, and on demand for testing.
+   * Done both at Catalog client startup, and on demand for testing.
    */
   @Override
   public void resync()

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java

@@ -26,7 +26,6 @@
 import org.apache.druid.catalog.model.TableDefnRegistry;
 import org.apache.druid.catalog.model.TableId;
 import org.apache.druid.catalog.model.TableMetadata;
-import org.apache.druid.catalog.sync.MetadataCatalog.CatalogSource;
 import org.apache.druid.client.coordinator.Coordinator;
 import org.apache.druid.discovery.DruidLeaderClient;
 import org.apache.druid.guice.annotations.Json;
@@ -41,7 +40,6 @@
 
 import javax.inject.Inject;
 import javax.ws.rs.core.MediaType;
-
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClientConfig.java

@@ -0,0 +1,71 @@
+/*
+ * 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.catalog.sync;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.concurrent.TimeUnit;
+
+public class CatalogClientConfig
+{
+  private static final long DEFAULT_POLLING_PERIOD = TimeUnit.MINUTES.toMillis(1);
+  private static final long DEFAULT_MAX_RANDOM_DELAY = TimeUnit.SECONDS.toMillis(10);
+  private static final int DEFAULT_MAX_SYNC_RETRIES = 5;
+
+  @JsonProperty
+  private final long pollingPeriod;
+
+  @JsonProperty
+  private final long maxRandomDelay;
+
+  @JsonProperty
+  private final int maxSyncRetries;
+
+  @JsonCreator
+  public CatalogClientConfig(
+      @JsonProperty("pollingPeriod") Long pollingPeriod,
+      @JsonProperty("maxRandomDelay") Long maxRandomDelay,
+      @JsonProperty("maxSyncRetries") Integer maxSyncRetries
+  )
+  {
+    this.pollingPeriod = pollingPeriod == null ? DEFAULT_POLLING_PERIOD : pollingPeriod;
+    this.maxRandomDelay = maxRandomDelay == null ? DEFAULT_MAX_RANDOM_DELAY : maxRandomDelay;
+    this.maxSyncRetries = maxSyncRetries == null ? DEFAULT_MAX_SYNC_RETRIES : maxSyncRetries;
+  }
+
+  @JsonProperty
+  public long getPollingPeriod()
+  {
+    return pollingPeriod;
+  }
+
+  @JsonProperty
+  public long getMaxRandomDelay()
+  {
+    return maxRandomDelay;
+  }
+
+  @JsonProperty
+  public int getMaxSyncRetries()
+  {
+    return maxSyncRetries;
+  }
+}

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogSource.java

@@ -0,0 +1,35 @@
+/*
+ * 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.catalog.sync;
+
+import org.apache.druid.catalog.model.ResolvedTable;
+import org.apache.druid.catalog.model.TableId;
+import org.apache.druid.catalog.model.TableMetadata;
+
+import java.util.List;
+
+public interface CatalogSource
+{
+  List<TableMetadata> tablesForSchema(String dbSchema);
+
+  TableMetadata table(TableId id);
+
+  ResolvedTable resolveTable(TableId id);
+}

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogUpdateNotifier.java

@@ -20,6 +20,7 @@
 package org.apache.druid.catalog.sync;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
 import org.apache.druid.catalog.http.CatalogListenerResource;
 import org.apache.druid.catalog.storage.CatalogStorage;
 import org.apache.druid.catalog.sync.RestUpdateSender.RestSender;
@@ -37,8 +38,6 @@
 import org.joda.time.Duration;
 
 import javax.inject.Inject;
-
-import java.util.Collections;
 import java.util.function.Supplier;
 
 /**
@@ -68,7 +67,7 @@ public CatalogUpdateNotifier(
     long timeoutMs = TIMEOUT_MS;
     this.smileMapper = smileMapper;
     Supplier<Iterable<DruidNode>> nodeSupplier = new ListeningNodeSupplier(
-        Collections.singletonList(NodeRole.BROKER),
+        ImmutableList.of(NodeRole.BROKER, NodeRole.OVERLORD),
         discoveryProvider);
     RestSender restSender = RestUpdateSender.httpClientSender(httpClient, Duration.millis(timeoutMs));
     RestUpdateSender sender = new RestUpdateSender(