[PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei opened a new pull request, #15469:
URL: https://github.com/apache/druid/pull/15469

   This PR contains a portion of the changes from the inactive draft PR for integrating the catalog with the Calcite planner https://github.com/apache/druid/pull/13686 from @paul-rogers, allowing the datasource table schemas defined in the catalog to be synced to the broker.
   
   This does not include the MSQ INSERT and REPLACE validation using the catalog functionality from https://github.com/apache/druid/pull/13686, this PR is just the portion of changes needed to sync the queryable schemas to the broker.
   
   This PR has:
   
   - [x] been self-reviewed.
   - [ ] added documentation for new or modified features or behaviors.
   - [ ] a release note entry in the PR description.
   - [x] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
   - [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
   - [x] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
   - [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
   - [ ] added integration tests.
   - [x] been tested in a test Druid cluster.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1419713185


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override
+  public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+  {
+    DatasourceFacade dsSpec = datasourceSpec(name);
+
+    // No catalog metadata. If there is no physical metadata, then the
+    // datasource does not exist. Else, if there is physical metadata, the
+    // datasource is based entirely on the physical information.
+    if (dsSpec == null) {
+      return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+    }
+    if (dsMetadata == null) {
+      // Datasource exists only in the catalog: no physical segments.
+      return emptyDatasource(name, dsSpec);
+    } else {
+      // Datasource exists as both segments and a catalog entry.
+      return mergeDatasource(dsMetadata, dsSpec);
+    }
+  }
+
+  private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+  {
+    RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    boolean hasTime = false;
+    for (ColumnSpec col : dsSpec.columns()) {
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+      if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+        hasTime = true;
+      }
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+    if (!hasTime) {
+      columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+          Columns.TIME_COLUMN,
+          ColumnType.LONG
+      ));
+      builder = RowSignature.builder()
+          .add(Columns.TIME_COLUMN, ColumnType.LONG)
+          .addAll(builder.build());
+    }
+
+    final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+          new TableDataSource(name),
+          builder.build(),
+          false, // Cannot join to an empty table
+          false // Cannot broadcast an empty table
+    );
+    return new DatasourceTable(
+        mergedMetadata.getRowSignature(),
+        mergedMetadata,
+        new EffectiveMetadata(dsSpec, columns, true)
+    );
+  }
+
+  private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+  {
+    ColumnType type = Columns.druidType(col);
+    if (type != null) {
+      // Use the type that the user provided.
+    } else if (physicalType == null) {
+      // Corner case: the user has defined a column in the catalog, has
+      // not specified a type (meaning the user wants Druid to decide), but
+      // there is no data at this moment. Guess String as the type for the
+      // null values. If new segments appear between now and execution, we'll
+      // convert the values to string, which is always safe.
+      type = ColumnType.STRING;
+    } else {
+      type = physicalType;
+    }
+    return new EffectiveColumnMetadata(col.name(), type);
+  }
+
+  private DruidTable mergeDatasource(
+      final PhysicalDatasourceMetadata dsMetadata,
+      final DatasourceFacade dsSpec)
+  {
+    final RowSignature physicalSchema = dsMetadata.getRowSignature();
+    Set<String> physicalCols = new HashSet<>(physicalSchema.getColumnNames());
+
+    // Merge columns. All catalog-defined columns come first,
+    // in the order defined in the catalog.
+    final RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();

Review Comment:
   `columns` is just used a column name lookup, the order-preserving component in a `DatasourceTable` is the `RowSignature` which stores the columns in a list



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418487949


##########
server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java:
##########
@@ -38,10 +44,77 @@ public class DatasourceFacade extends TableFacade
 {
   private static final Logger LOG = new Logger(DatasourceFacade.class);
 
+  public static class ColumnFacade
+  {
+    public enum Kind
+    {
+      ANY,
+      TIME,
+      DIMENSION,
+      MEASURE
+    }
+
+    private final ColumnSpec spec;
+    private final String sqlType;
+
+    public ColumnFacade(ColumnSpec spec)
+    {
+      this.spec = spec;
+      if (Columns.isTimeColumn(spec.name()) && spec.dataType() == null) {
+        // For __time only, force a type if type is null.
+        this.sqlType = Columns.LONG;

Review Comment:
   Should this be `COLUMNS.SQL_BIGINT`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418469271


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -86,7 +86,7 @@ public CatalogClient(
   @Override
   public List<TableMetadata> tablesForSchema(String dbSchema)
   {
-    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{dbSchema}", dbSchema);
+    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{schema}", dbSchema);

Review Comment:
   might be better to use the constant `CatalogResource.SCHEMA_SYNC` here



##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -96,7 +96,7 @@ public List<TableMetadata> tablesForSchema(String dbSchema)
   @Override
   public TableMetadata table(TableId id)
   {
-    String url = StringUtils.replace(TABLE_SYNC_PATH, "{dbSchema}", id.schema());
+    String url = StringUtils.replace(TABLE_SYNC_PATH, "{schema}", id.schema());

Review Comment:
   ditto



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418501773


##########
server/src/main/java/org/apache/druid/catalog/model/table/ExternalTableDefn.java:
##########
@@ -205,6 +205,11 @@ public class ExternalTableDefn extends TableDefn
    */
   public static final String TABLE_TYPE = "extern";
 
+  /**
+   * Column type for external tables.
+   */
+  public static final String EXTERNAL_COLUMN_TYPE = "extern";

Review Comment:
   What is this used for?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418478640


##########
server/src/main/java/org/apache/druid/catalog/model/ColumnSpec.java:
##########
@@ -49,14 +48,10 @@ public class ColumnSpec
   private final String name;
 
   /**
-   * The data type of the column expressed as a supported SQL type. The data type here must
-   * directly match a Druid storage type. So, {@code BIGINT} for {code long}, say.
-   * This usage does not support Druid's usual "fudging": one cannot use {@code INTEGER}
-   * to mean {@code long}. The type will likely encode complex and aggregation types
-   * in the future, though that is not yet supported. The set of valid mappings is
-   * defined in the {@link Columns} class.
+   * The data type of the column expressed as a supported Druid type. The data type here must
+   * directly match a Druid storage type.
    */
-  private final String sqlType;
+  private final String dataType;

Review Comment:
   What is the reasoning behind changing from sql type to druid type here?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1419803603


##########
sql/src/main/java/org/apache/druid/sql/calcite/table/DatasourceTable.java:
##########
@@ -115,14 +137,107 @@ public String toString()
     }
   }
 
+  public static class EffectiveColumnMetadata
+  {
+    protected final String name;
+    protected final ColumnType type;
+
+    public EffectiveColumnMetadata(String name, ColumnType type)
+    {
+      this.name = name;
+      this.type = type;
+    }
+
+    public String name()
+    {
+      return name;
+    }
+
+    public ColumnType druidType()
+    {
+      return type;
+    }
+
+    public static EffectiveColumnMetadata fromPhysical(String name, ColumnType type)
+    {
+      return new EffectiveColumnMetadata(name, type);
+    }
+
+    @Override
+    public String toString()
+    {
+      return "Column{" +
+             "name=" + name +
+             ", type=" + type.asTypeString() +
+             "}";
+    }
+  }
+
+  public static class EffectiveMetadata
+  {
+    private final DatasourceFacade catalogMetadata;
+    private final boolean isEmpty;
+    private final Map<String, EffectiveColumnMetadata> columns;
+
+    public EffectiveMetadata(
+        final DatasourceFacade catalogMetadata,
+        final Map<String, EffectiveColumnMetadata> columns,
+        final boolean isEmpty
+    )
+    {
+      this.catalogMetadata = catalogMetadata;
+      this.isEmpty = isEmpty;
+      this.columns = columns;
+    }
+
+    public DatasourceFacade catalogMetadata()
+    {
+      return catalogMetadata;
+    }
+
+    public EffectiveColumnMetadata column(String name)
+    {
+      return columns.get(name);
+    }
+
+    public boolean isEmpty()

Review Comment:
   Yes, it's set to true when there are no physical segments



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1416342166


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver

Review Comment:
   javadoc



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418500913


##########
server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java:
##########
@@ -68,16 +68,6 @@ public class DatasourceDefn extends TableDefn
    */
   public static final String HIDDEN_COLUMNS_PROPERTY = "hiddenColumns";
 
-  /**
-   * By default: columns are optional hints. If a datasource has columns defined,

Review Comment:
   Why does this description no longer apply?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei merged PR #15469:
URL: https://github.com/apache/druid/pull/15469


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423022496


##########
server/src/main/java/org/apache/druid/catalog/model/Columns.java:
##########
@@ -34,58 +35,72 @@ public class Columns
 {
   public static final String TIME_COLUMN = "__time";
 
-  public static final String VARCHAR = "VARCHAR";
-  public static final String BIGINT = "BIGINT";
-  public static final String FLOAT = "FLOAT";
-  public static final String DOUBLE = "DOUBLE";
-  public static final String TIMESTAMP = "TIMESTAMP";
+  public static final String STRING = ValueType.STRING.name();
+  public static final String LONG = ValueType.LONG.name();
+  public static final String FLOAT = ValueType.FLOAT.name();
+  public static final String DOUBLE = ValueType.DOUBLE.name();
+
+  public static final String SQL_VARCHAR = "VARCHAR";
+  public static final String SQL_BIGINT = "BIGINT";
+  public static final String SQL_FLOAT = "FLOAT";
+  public static final String SQL_DOUBLE = "DOUBLE";
+  public static final String SQL_TIMESTAMP = "TIMESTAMP";
 
   public static final Set<String> NUMERIC_TYPES =
-      ImmutableSet.of(BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(LONG, FLOAT, DOUBLE);
   public static final Set<String> SCALAR_TYPES =
-      ImmutableSet.of(TIMESTAMP, VARCHAR, BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(STRING, LONG, FLOAT, DOUBLE);
 
   public static final Map<String, ColumnType> SQL_TO_DRUID_TYPES =
       new ImmutableMap.Builder<String, ColumnType>()
-        .put(TIMESTAMP, ColumnType.LONG)
-        .put(BIGINT, ColumnType.LONG)
-        .put(FLOAT, ColumnType.FLOAT)
-        .put(DOUBLE, ColumnType.DOUBLE)
-        .put(VARCHAR, ColumnType.STRING)
-        .build();
+          .put(SQL_BIGINT, ColumnType.LONG)
+          .put(SQL_VARCHAR, ColumnType.STRING)

Review Comment:
   This part wasn't being used actually right now, removed it



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423022989


##########
server/src/main/java/org/apache/druid/catalog/model/table/ExternalTableDefn.java:
##########
@@ -205,6 +205,11 @@ public class ExternalTableDefn extends TableDefn
    */
   public static final String TABLE_TYPE = "extern";
 
+  /**
+   * Column type for external tables.
+   */
+  public static final String EXTERNAL_COLUMN_TYPE = "extern";

Review Comment:
   This is in another part of the original PR related to external table handling, it's not used currently but future changes will



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423017826


##########
server/src/main/java/org/apache/druid/catalog/model/Columns.java:
##########
@@ -34,58 +35,72 @@ public class Columns
 {
   public static final String TIME_COLUMN = "__time";
 
-  public static final String VARCHAR = "VARCHAR";
-  public static final String BIGINT = "BIGINT";
-  public static final String FLOAT = "FLOAT";
-  public static final String DOUBLE = "DOUBLE";
-  public static final String TIMESTAMP = "TIMESTAMP";
+  public static final String STRING = ValueType.STRING.name();
+  public static final String LONG = ValueType.LONG.name();
+  public static final String FLOAT = ValueType.FLOAT.name();
+  public static final String DOUBLE = ValueType.DOUBLE.name();
+
+  public static final String SQL_VARCHAR = "VARCHAR";
+  public static final String SQL_BIGINT = "BIGINT";
+  public static final String SQL_FLOAT = "FLOAT";
+  public static final String SQL_DOUBLE = "DOUBLE";
+  public static final String SQL_TIMESTAMP = "TIMESTAMP";
 
   public static final Set<String> NUMERIC_TYPES =
-      ImmutableSet.of(BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(LONG, FLOAT, DOUBLE);
   public static final Set<String> SCALAR_TYPES =
-      ImmutableSet.of(TIMESTAMP, VARCHAR, BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(STRING, LONG, FLOAT, DOUBLE);
 
   public static final Map<String, ColumnType> SQL_TO_DRUID_TYPES =
       new ImmutableMap.Builder<String, ColumnType>()
-        .put(TIMESTAMP, ColumnType.LONG)
-        .put(BIGINT, ColumnType.LONG)
-        .put(FLOAT, ColumnType.FLOAT)
-        .put(DOUBLE, ColumnType.DOUBLE)
-        .put(VARCHAR, ColumnType.STRING)
-        .build();
+          .put(SQL_BIGINT, ColumnType.LONG)
+          .put(SQL_VARCHAR, ColumnType.STRING)
+          .build();
 
   public static final Map<ColumnType, String> DRUID_TO_SQL_TYPES =
       new ImmutableMap.Builder<ColumnType, String>()
-      .put(ColumnType.LONG, BIGINT)
-      .put(ColumnType.FLOAT, FLOAT)
-      .put(ColumnType.DOUBLE, DOUBLE)
-      .put(ColumnType.STRING, VARCHAR)
+      .put(ColumnType.LONG, SQL_BIGINT)
+      .put(ColumnType.FLOAT, SQL_FLOAT)
+      .put(ColumnType.DOUBLE, SQL_DOUBLE)
+      .put(ColumnType.STRING, SQL_VARCHAR)
       .build();
 
   private Columns()
   {
   }
 
-  public static boolean isTimestamp(String type)
-  {
-    return TIMESTAMP.equalsIgnoreCase(type.trim());
-  }
-
   public static boolean isScalar(String type)
   {
     return SCALAR_TYPES.contains(StringUtils.toUpperCase(type.trim()));
   }
 
-  public static ColumnType druidType(String sqlType)
+  public static ColumnType druidType(ColumnSpec spec)
   {
-    if (sqlType == null) {
+    if (isTimeColumn(spec.name())) {
+      return ColumnType.LONG;
+    }
+    String dataType = spec.dataType();
+    if (dataType == null) {
       return null;
     }
-    ColumnType druidType = SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(sqlType));
+    ColumnType druidType = SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(dataType));

Review Comment:
   It checks if it's a SQL type first and attempts to convert, and then does `ColumnType.fromString(dataType)` if not to treat it as a Druid native type



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423019096


##########
server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java:
##########
@@ -38,10 +44,77 @@ public class DatasourceFacade extends TableFacade
 {
   private static final Logger LOG = new Logger(DatasourceFacade.class);
 
+  public static class ColumnFacade
+  {
+    public enum Kind
+    {
+      ANY,
+      TIME,
+      DIMENSION,
+      MEASURE
+    }
+
+    private final ColumnSpec spec;
+    private final String sqlType;
+
+    public ColumnFacade(ColumnSpec spec)
+    {
+      this.spec = spec;
+      if (Columns.isTimeColumn(spec.name()) && spec.dataType() == null) {
+        // For __time only, force a type if type is null.
+        this.sqlType = Columns.LONG;

Review Comment:
   No, this is using the native type



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418484305


##########
server/src/main/java/org/apache/druid/catalog/model/Columns.java:
##########
@@ -34,58 +35,72 @@ public class Columns
 {
   public static final String TIME_COLUMN = "__time";
 
-  public static final String VARCHAR = "VARCHAR";
-  public static final String BIGINT = "BIGINT";
-  public static final String FLOAT = "FLOAT";
-  public static final String DOUBLE = "DOUBLE";
-  public static final String TIMESTAMP = "TIMESTAMP";
+  public static final String STRING = ValueType.STRING.name();
+  public static final String LONG = ValueType.LONG.name();
+  public static final String FLOAT = ValueType.FLOAT.name();
+  public static final String DOUBLE = ValueType.DOUBLE.name();
+
+  public static final String SQL_VARCHAR = "VARCHAR";
+  public static final String SQL_BIGINT = "BIGINT";
+  public static final String SQL_FLOAT = "FLOAT";
+  public static final String SQL_DOUBLE = "DOUBLE";
+  public static final String SQL_TIMESTAMP = "TIMESTAMP";
 
   public static final Set<String> NUMERIC_TYPES =
-      ImmutableSet.of(BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(LONG, FLOAT, DOUBLE);
   public static final Set<String> SCALAR_TYPES =
-      ImmutableSet.of(TIMESTAMP, VARCHAR, BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(STRING, LONG, FLOAT, DOUBLE);
 
   public static final Map<String, ColumnType> SQL_TO_DRUID_TYPES =
       new ImmutableMap.Builder<String, ColumnType>()
-        .put(TIMESTAMP, ColumnType.LONG)
-        .put(BIGINT, ColumnType.LONG)
-        .put(FLOAT, ColumnType.FLOAT)
-        .put(DOUBLE, ColumnType.DOUBLE)
-        .put(VARCHAR, ColumnType.STRING)
-        .build();
+          .put(SQL_BIGINT, ColumnType.LONG)
+          .put(SQL_VARCHAR, ColumnType.STRING)
+          .build();
 
   public static final Map<ColumnType, String> DRUID_TO_SQL_TYPES =
       new ImmutableMap.Builder<ColumnType, String>()
-      .put(ColumnType.LONG, BIGINT)
-      .put(ColumnType.FLOAT, FLOAT)
-      .put(ColumnType.DOUBLE, DOUBLE)
-      .put(ColumnType.STRING, VARCHAR)
+      .put(ColumnType.LONG, SQL_BIGINT)
+      .put(ColumnType.FLOAT, SQL_FLOAT)
+      .put(ColumnType.DOUBLE, SQL_DOUBLE)
+      .put(ColumnType.STRING, SQL_VARCHAR)
       .build();
 
   private Columns()
   {
   }
 
-  public static boolean isTimestamp(String type)
-  {
-    return TIMESTAMP.equalsIgnoreCase(type.trim());
-  }
-
   public static boolean isScalar(String type)
   {
     return SCALAR_TYPES.contains(StringUtils.toUpperCase(type.trim()));
   }
 
-  public static ColumnType druidType(String sqlType)
+  public static ColumnType druidType(ColumnSpec spec)
   {
-    if (sqlType == null) {
+    if (isTimeColumn(spec.name())) {
+      return ColumnType.LONG;
+    }
+    String dataType = spec.dataType();
+    if (dataType == null) {
       return null;
     }
-    ColumnType druidType = SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(sqlType));
+    ColumnType druidType = SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(dataType));

Review Comment:
   It looks like we changed the dataType of columnSpec to be the druid type instead of the sql typ now, so should we take the type as is now instead of looking up mapping here?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1416343243


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override

Review Comment:
   `@Nullable`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["github-advanced-security[bot] (via GitHub)" <gi...@apache.org>]

github-advanced-security[bot] commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1411662305


##########
extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogQueryTest.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.sql;
+
+import org.apache.druid.catalog.CatalogException;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.table.TableBuilder;
+import org.apache.druid.catalog.storage.CatalogStorage;
+import org.apache.druid.catalog.storage.CatalogTests;
+import org.apache.druid.catalog.sync.CachedMetadataCatalog;
+import org.apache.druid.catalog.sync.MetadataCatalog;
+import org.apache.druid.metadata.TestDerbyConnector;
+import org.apache.druid.sql.calcite.BaseCalciteQueryTest;
+import org.apache.druid.sql.calcite.SqlSchema;
+import org.apache.druid.sql.calcite.util.SqlTestFramework;
+import org.apache.druid.sql.calcite.util.SqlTestFramework.Builder;
+import org.junit.After;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.Assert.fail;
+
+public class CatalogQueryTest extends BaseCalciteQueryTest
+{
+  @Rule
+  public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule = new TestDerbyConnector.DerbyConnectorRule();
+
+  private CatalogTests.DbFixture dbFixture;
+  private CatalogStorage storage;
+
+  @Test
+  public void testCatalogSchema()
+  {
+    SqlSchema schema = SqlSchema.builder()
+        .column("__time", "TIMESTAMP(3) NOT NULL")
+        .column("extra1", "VARCHAR")
+        .column("dim2", "VARCHAR")
+        .column("dim1", "VARCHAR")
+        .column("cnt", "BIGINT NOT NULL")
+        .column("m1", "DOUBLE NOT NULL")
+        .column("extra2", "BIGINT NOT NULL")
+        .column("extra3", "VARCHAR")
+        .column("m2", "DOUBLE NOT NULL")
+        .build();

Review Comment:
   ## Unread local variable
   
   Variable 'SqlSchema schema' is never read.
   
   [Show more details](https://github.com/apache/druid/security/code-scanning/6018)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1416357939


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override
+  public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+  {
+    DatasourceFacade dsSpec = datasourceSpec(name);
+
+    // No catalog metadata. If there is no physical metadata, then the
+    // datasource does not exist. Else, if there is physical metadata, the
+    // datasource is based entirely on the physical information.
+    if (dsSpec == null) {
+      return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+    }
+    if (dsMetadata == null) {
+      // Datasource exists only in the catalog: no physical segments.
+      return emptyDatasource(name, dsSpec);
+    } else {
+      // Datasource exists as both segments and a catalog entry.
+      return mergeDatasource(dsMetadata, dsSpec);
+    }
+  }
+
+  private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+  {
+    RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    boolean hasTime = false;
+    for (ColumnSpec col : dsSpec.columns()) {
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+      if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+        hasTime = true;
+      }
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+    if (!hasTime) {
+      columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+          Columns.TIME_COLUMN,
+          ColumnType.LONG
+      ));
+      builder = RowSignature.builder()
+          .add(Columns.TIME_COLUMN, ColumnType.LONG)
+          .addAll(builder.build());
+    }
+
+    final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+          new TableDataSource(name),
+          builder.build(),
+          false, // Cannot join to an empty table
+          false // Cannot broadcast an empty table
+    );
+    return new DatasourceTable(
+        mergedMetadata.getRowSignature(),
+        mergedMetadata,
+        new EffectiveMetadata(dsSpec, columns, true)
+    );
+  }
+
+  private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+  {
+    ColumnType type = Columns.druidType(col);
+    if (type != null) {
+      // Use the type that the user provided.
+    } else if (physicalType == null) {
+      // Corner case: the user has defined a column in the catalog, has
+      // not specified a type (meaning the user wants Druid to decide), but
+      // there is no data at this moment. Guess String as the type for the
+      // null values. If new segments appear between now and execution, we'll
+      // convert the values to string, which is always safe.
+      type = ColumnType.STRING;
+    } else {
+      type = physicalType;
+    }
+    return new EffectiveColumnMetadata(col.name(), type);
+  }
+
+  private DruidTable mergeDatasource(
+      final PhysicalDatasourceMetadata dsMetadata,
+      final DatasourceFacade dsSpec)
+  {
+    final RowSignature physicalSchema = dsMetadata.getRowSignature();
+    Set<String> physicalCols = new HashSet<>(physicalSchema.getColumnNames());
+
+    // Merge columns. All catalog-defined columns come first,
+    // in the order defined in the catalog.
+    final RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();

Review Comment:
   Does the insertion of keys, values into hashMap preserve order that it was added? I assume that the order of columns  of the effectiveMetadata is important?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418470458


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -86,7 +86,7 @@ public CatalogClient(
   @Override
   public List<TableMetadata> tablesForSchema(String dbSchema)
   {
-    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{dbSchema}", dbSchema);
+    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{schema}", dbSchema);

Review Comment:
   nit: Good catch. should we move `"{schema}"` to a constant here?



##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -96,7 +96,7 @@ public List<TableMetadata> tablesForSchema(String dbSchema)
   @Override
   public TableMetadata table(TableId id)
   {
-    String url = StringUtils.replace(TABLE_SYNC_PATH, "{dbSchema}", id.schema());
+    String url = StringUtils.replace(TABLE_SYNC_PATH, "{schema}", id.schema());

Review Comment:
   ditto



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1416354340


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override
+  public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+  {
+    DatasourceFacade dsSpec = datasourceSpec(name);
+
+    // No catalog metadata. If there is no physical metadata, then the
+    // datasource does not exist. Else, if there is physical metadata, the
+    // datasource is based entirely on the physical information.
+    if (dsSpec == null) {
+      return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+    }
+    if (dsMetadata == null) {
+      // Datasource exists only in the catalog: no physical segments.
+      return emptyDatasource(name, dsSpec);
+    } else {
+      // Datasource exists as both segments and a catalog entry.
+      return mergeDatasource(dsMetadata, dsSpec);
+    }
+  }
+
+  private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+  {
+    RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    boolean hasTime = false;
+    for (ColumnSpec col : dsSpec.columns()) {
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+      if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+        hasTime = true;
+      }
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+    if (!hasTime) {
+      columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+          Columns.TIME_COLUMN,
+          ColumnType.LONG
+      ));
+      builder = RowSignature.builder()
+          .add(Columns.TIME_COLUMN, ColumnType.LONG)
+          .addAll(builder.build());
+    }
+
+    final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+          new TableDataSource(name),
+          builder.build(),
+          false, // Cannot join to an empty table
+          false // Cannot broadcast an empty table
+    );
+    return new DatasourceTable(
+        mergedMetadata.getRowSignature(),
+        mergedMetadata,
+        new EffectiveMetadata(dsSpec, columns, true)
+    );
+  }
+
+  private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+  {
+    ColumnType type = Columns.druidType(col);
+    if (type != null) {
+      // Use the type that the user provided.
+    } else if (physicalType == null) {
+      // Corner case: the user has defined a column in the catalog, has
+      // not specified a type (meaning the user wants Druid to decide), but
+      // there is no data at this moment. Guess String as the type for the
+      // null values. If new segments appear between now and execution, we'll
+      // convert the values to string, which is always safe.
+      type = ColumnType.STRING;
+    } else {
+      type = physicalType;
+    }
+    return new EffectiveColumnMetadata(col.name(), type);
+  }
+
+  private DruidTable mergeDatasource(
+      final PhysicalDatasourceMetadata dsMetadata,
+      final DatasourceFacade dsSpec)
+  {
+    final RowSignature physicalSchema = dsMetadata.getRowSignature();
+    Set<String> physicalCols = new HashSet<>(physicalSchema.getColumnNames());
+
+    // Merge columns. All catalog-defined columns come first,
+    // in the order defined in the catalog.
+    final RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    for (ColumnSpec col : dsSpec.columns()) {
+      ColumnType physicalType = null;
+      if (physicalCols.remove(col.name())) {
+        physicalType = dsMetadata.getRowSignature().getColumnType(col.name()).get();
+      }
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, physicalType);
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+
+    // Mark any hidden columns. Assumes that the hidden columns are a disjoint set

Review Comment:
   does anything enforce this disjoint property? What if hidden columns are also defined?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423017039


##########
server/src/main/java/org/apache/druid/catalog/model/ColumnSpec.java:
##########
@@ -49,14 +48,10 @@ public class ColumnSpec
   private final String name;
 
   /**
-   * The data type of the column expressed as a supported SQL type. The data type here must
-   * directly match a Druid storage type. So, {@code BIGINT} for {code long}, say.
-   * This usage does not support Druid's usual "fudging": one cannot use {@code INTEGER}
-   * to mean {@code long}. The type will likely encode complex and aggregation types
-   * in the future, though that is not yet supported. The set of valid mappings is
-   * defined in the {@link Columns} class.
+   * The data type of the column expressed as a supported Druid type. The data type here must
+   * directly match a Druid storage type.
    */
-  private final String sqlType;
+  private final String dataType;

Review Comment:
   It now accepts both SQL and Druid native type strings in the catalog definition



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418481065


##########
server/src/main/java/org/apache/druid/catalog/model/Columns.java:
##########
@@ -34,58 +35,72 @@ public class Columns
 {
   public static final String TIME_COLUMN = "__time";
 
-  public static final String VARCHAR = "VARCHAR";
-  public static final String BIGINT = "BIGINT";
-  public static final String FLOAT = "FLOAT";
-  public static final String DOUBLE = "DOUBLE";
-  public static final String TIMESTAMP = "TIMESTAMP";
+  public static final String STRING = ValueType.STRING.name();
+  public static final String LONG = ValueType.LONG.name();
+  public static final String FLOAT = ValueType.FLOAT.name();
+  public static final String DOUBLE = ValueType.DOUBLE.name();
+
+  public static final String SQL_VARCHAR = "VARCHAR";
+  public static final String SQL_BIGINT = "BIGINT";
+  public static final String SQL_FLOAT = "FLOAT";
+  public static final String SQL_DOUBLE = "DOUBLE";
+  public static final String SQL_TIMESTAMP = "TIMESTAMP";
 
   public static final Set<String> NUMERIC_TYPES =
-      ImmutableSet.of(BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(LONG, FLOAT, DOUBLE);
   public static final Set<String> SCALAR_TYPES =
-      ImmutableSet.of(TIMESTAMP, VARCHAR, BIGINT, FLOAT, DOUBLE);
+      ImmutableSet.of(STRING, LONG, FLOAT, DOUBLE);
 
   public static final Map<String, ColumnType> SQL_TO_DRUID_TYPES =
       new ImmutableMap.Builder<String, ColumnType>()
-        .put(TIMESTAMP, ColumnType.LONG)
-        .put(BIGINT, ColumnType.LONG)
-        .put(FLOAT, ColumnType.FLOAT)
-        .put(DOUBLE, ColumnType.DOUBLE)
-        .put(VARCHAR, ColumnType.STRING)
-        .build();
+          .put(SQL_BIGINT, ColumnType.LONG)
+          .put(SQL_VARCHAR, ColumnType.STRING)

Review Comment:
   Do you mean to remove mappings for `SQL_FLOAT` and `SQL_DOUBLE` and `TIMESTAMP` here?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418473717


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogUpdateReceiver.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.guice.ManageLifecycle;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+
+import javax.inject.Inject;
+
+/**
+ * Receiver which runs in the Broker to listen for catalog updates from the
+ * Coordinator. To prevent slowing initial queries, this class loads the
+ * current catalog contents into the local cache on lifecycle start, which
+ * avoids the on-demand reads that would otherwise occur. After the first load,
+ * events from the Coordinator keep the local cache evergreen.

Review Comment:
   Where are events from the coordinator pushed after this receiver loads on lifecycle start? May be good to note that here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418469271


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -86,7 +86,7 @@ public CatalogClient(
   @Override
   public List<TableMetadata> tablesForSchema(String dbSchema)
   {
-    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{dbSchema}", dbSchema);
+    String url = StringUtils.replace(SCHEMA_SYNC_PATH, "{schema}", dbSchema);

Review Comment:
   might be better to use the constant `CatalogResource.SCHEMA_SYNC` here



##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogClient.java:
##########
@@ -96,7 +96,7 @@ public List<TableMetadata> tablesForSchema(String dbSchema)
   @Override
   public TableMetadata table(TableId id)
   {
-    String url = StringUtils.replace(TABLE_SYNC_PATH, "{dbSchema}", id.schema());
+    String url = StringUtils.replace(TABLE_SYNC_PATH, "{schema}", id.schema());

Review Comment:
   ditto



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["zachjsh (via GitHub)" <gi...@apache.org>]

zachjsh commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1418509963


##########
sql/src/main/java/org/apache/druid/sql/calcite/table/DatasourceTable.java:
##########
@@ -115,14 +137,107 @@ public String toString()
     }
   }
 
+  public static class EffectiveColumnMetadata
+  {
+    protected final String name;
+    protected final ColumnType type;
+
+    public EffectiveColumnMetadata(String name, ColumnType type)
+    {
+      this.name = name;
+      this.type = type;
+    }
+
+    public String name()
+    {
+      return name;
+    }
+
+    public ColumnType druidType()
+    {
+      return type;
+    }
+
+    public static EffectiveColumnMetadata fromPhysical(String name, ColumnType type)
+    {
+      return new EffectiveColumnMetadata(name, type);
+    }
+
+    @Override
+    public String toString()
+    {
+      return "Column{" +
+             "name=" + name +
+             ", type=" + type.asTypeString() +
+             "}";
+    }
+  }
+
+  public static class EffectiveMetadata
+  {
+    private final DatasourceFacade catalogMetadata;
+    private final boolean isEmpty;
+    private final Map<String, EffectiveColumnMetadata> columns;
+
+    public EffectiveMetadata(
+        final DatasourceFacade catalogMetadata,
+        final Map<String, EffectiveColumnMetadata> columns,
+        final boolean isEmpty
+    )
+    {
+      this.catalogMetadata = catalogMetadata;
+      this.isEmpty = isEmpty;
+      this.columns = columns;
+    }
+
+    public DatasourceFacade catalogMetadata()
+    {
+      return catalogMetadata;
+    }
+
+    public EffectiveColumnMetadata column(String name)
+    {
+      return columns.get(name);
+    }
+
+    public boolean isEmpty()

Review Comment:
   Does isEmpty signify whether or not the table had any data in it or not?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1419713185


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override
+  public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+  {
+    DatasourceFacade dsSpec = datasourceSpec(name);
+
+    // No catalog metadata. If there is no physical metadata, then the
+    // datasource does not exist. Else, if there is physical metadata, the
+    // datasource is based entirely on the physical information.
+    if (dsSpec == null) {
+      return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+    }
+    if (dsMetadata == null) {
+      // Datasource exists only in the catalog: no physical segments.
+      return emptyDatasource(name, dsSpec);
+    } else {
+      // Datasource exists as both segments and a catalog entry.
+      return mergeDatasource(dsMetadata, dsSpec);
+    }
+  }
+
+  private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+  {
+    RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    boolean hasTime = false;
+    for (ColumnSpec col : dsSpec.columns()) {
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+      if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+        hasTime = true;
+      }
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+    if (!hasTime) {
+      columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+          Columns.TIME_COLUMN,
+          ColumnType.LONG
+      ));
+      builder = RowSignature.builder()
+          .add(Columns.TIME_COLUMN, ColumnType.LONG)
+          .addAll(builder.build());
+    }
+
+    final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+          new TableDataSource(name),
+          builder.build(),
+          false, // Cannot join to an empty table
+          false // Cannot broadcast an empty table
+    );
+    return new DatasourceTable(
+        mergedMetadata.getRowSignature(),
+        mergedMetadata,
+        new EffectiveMetadata(dsSpec, columns, true)
+    );
+  }
+
+  private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+  {
+    ColumnType type = Columns.druidType(col);
+    if (type != null) {
+      // Use the type that the user provided.
+    } else if (physicalType == null) {
+      // Corner case: the user has defined a column in the catalog, has
+      // not specified a type (meaning the user wants Druid to decide), but
+      // there is no data at this moment. Guess String as the type for the
+      // null values. If new segments appear between now and execution, we'll
+      // convert the values to string, which is always safe.
+      type = ColumnType.STRING;
+    } else {
+      type = physicalType;
+    }
+    return new EffectiveColumnMetadata(col.name(), type);
+  }
+
+  private DruidTable mergeDatasource(
+      final PhysicalDatasourceMetadata dsMetadata,
+      final DatasourceFacade dsSpec)
+  {
+    final RowSignature physicalSchema = dsMetadata.getRowSignature();
+    Set<String> physicalCols = new HashSet<>(physicalSchema.getColumnNames());
+
+    // Merge columns. All catalog-defined columns come first,
+    // in the order defined in the catalog.
+    final RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();

Review Comment:
   `columns` is just used as column name lookup, the order-preserving component in a `DatasourceTable` is the `RowSignature` which stores the columns in a list



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1419729223


##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.sql;
+
+import com.google.common.collect.ImmutableSet;
+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;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+  private final MetadataCatalog catalog;
+
+  @Inject
+  public LiveCatalogResolver(final MetadataCatalog catalog)
+  {
+    this.catalog = catalog;
+  }
+
+  private DatasourceFacade datasourceSpec(String name)
+  {
+    TableId tableId = TableId.datasource(name);
+    ResolvedTable table = catalog.resolveTable(tableId);
+    if (table == null) {
+      return null;
+    }
+    if (!DatasourceDefn.isDatasource(table)) {
+      return null;
+    }
+    return new DatasourceFacade(table);
+  }
+
+  /**
+   * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+   */
+  @Override
+  public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+  {
+    DatasourceFacade dsSpec = datasourceSpec(name);
+
+    // No catalog metadata. If there is no physical metadata, then the
+    // datasource does not exist. Else, if there is physical metadata, the
+    // datasource is based entirely on the physical information.
+    if (dsSpec == null) {
+      return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+    }
+    if (dsMetadata == null) {
+      // Datasource exists only in the catalog: no physical segments.
+      return emptyDatasource(name, dsSpec);
+    } else {
+      // Datasource exists as both segments and a catalog entry.
+      return mergeDatasource(dsMetadata, dsSpec);
+    }
+  }
+
+  private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+  {
+    RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    boolean hasTime = false;
+    for (ColumnSpec col : dsSpec.columns()) {
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+      if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+        hasTime = true;
+      }
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+    if (!hasTime) {
+      columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+          Columns.TIME_COLUMN,
+          ColumnType.LONG
+      ));
+      builder = RowSignature.builder()
+          .add(Columns.TIME_COLUMN, ColumnType.LONG)
+          .addAll(builder.build());
+    }
+
+    final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+          new TableDataSource(name),
+          builder.build(),
+          false, // Cannot join to an empty table
+          false // Cannot broadcast an empty table
+    );
+    return new DatasourceTable(
+        mergedMetadata.getRowSignature(),
+        mergedMetadata,
+        new EffectiveMetadata(dsSpec, columns, true)
+    );
+  }
+
+  private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+  {
+    ColumnType type = Columns.druidType(col);
+    if (type != null) {
+      // Use the type that the user provided.
+    } else if (physicalType == null) {
+      // Corner case: the user has defined a column in the catalog, has
+      // not specified a type (meaning the user wants Druid to decide), but
+      // there is no data at this moment. Guess String as the type for the
+      // null values. If new segments appear between now and execution, we'll
+      // convert the values to string, which is always safe.
+      type = ColumnType.STRING;
+    } else {
+      type = physicalType;
+    }
+    return new EffectiveColumnMetadata(col.name(), type);
+  }
+
+  private DruidTable mergeDatasource(
+      final PhysicalDatasourceMetadata dsMetadata,
+      final DatasourceFacade dsSpec)
+  {
+    final RowSignature physicalSchema = dsMetadata.getRowSignature();
+    Set<String> physicalCols = new HashSet<>(physicalSchema.getColumnNames());
+
+    // Merge columns. All catalog-defined columns come first,
+    // in the order defined in the catalog.
+    final RowSignature.Builder builder = RowSignature.builder();
+    Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+    for (ColumnSpec col : dsSpec.columns()) {
+      ColumnType physicalType = null;
+      if (physicalCols.remove(col.name())) {
+        physicalType = dsMetadata.getRowSignature().getColumnType(col.name()).get();
+      }
+      EffectiveColumnMetadata colMetadata = columnFromCatalog(col, physicalType);
+      builder.add(col.name(), colMetadata.druidType());
+      columns.put(col.name(), colMetadata);
+    }
+
+    // Mark any hidden columns. Assumes that the hidden columns are a disjoint set

Review Comment:
   There isn't enforcement of this right now, as-is if hidden columns overlap with the declared columns, the hiding would be ignored (the user would need to remove the column from the declared set to have the hiding take effect)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Allow broker to use catalog for datasource schemas for SQL queries (druid)

["jon-wei (via GitHub)" <gi...@apache.org>]

jon-wei commented on code in PR #15469:
URL: https://github.com/apache/druid/pull/15469#discussion_r1423020312


##########
server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java:
##########
@@ -68,16 +68,6 @@ public class DatasourceDefn extends TableDefn
    */
   public static final String HIDDEN_COLUMNS_PROPERTY = "hiddenColumns";
 
-  /**
-   * By default: columns are optional hints. If a datasource has columns defined,

Review Comment:
   I'm not sure why this was removed in Paul's original PR



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org