Merge branch 'integration' into feature/index-expansion-cache

pom.xml

@@ -67,7 +67,7 @@
         <version.datawave.common-utils>3.0.0</version.datawave.common-utils>
         <version.datawave.dictionary-api>4.0.1</version.datawave.dictionary-api>
         <version.datawave.mapreduce-query-api>1.0.0</version.datawave.mapreduce-query-api>
-        <version.datawave.metadata-utils>4.0.7</version.datawave.metadata-utils>
+        <version.datawave.metadata-utils>4.0.8</version.datawave.metadata-utils>
         <version.datawave.metrics-reporter>3.0.0</version.datawave.metrics-reporter>
         <version.datawave.query-api>1.0.0</version.datawave.query-api>
         <version.datawave.query-metric-api>4.0.7</version.datawave.query-metric-api>

warehouse/query-core/src/main/java/datawave/query/config/ShardQueryConfiguration.java

@@ -122,7 +122,6 @@ public class ShardQueryConfiguration extends GenericQueryConfiguration implement
     private boolean reduceQueryFieldsPerShard = false;
     private boolean reduceTypeMetadata = false;
     private boolean reduceTypeMetadataPerShard = false;
-    private boolean sequentialScheduler = false;
     private boolean collectTimingDetails = false;
     private boolean logTimingDetails = false;
     private boolean sendTimingToStatsd = true;
@@ -568,7 +567,6 @@ public void copyFrom(ShardQueryConfiguration other) {
         this.setRebuildDatatypeFilter(other.isRebuildDatatypeFilter());
         this.setRebuildDatatypeFilterPerShard(other.isRebuildDatatypeFilterPerShard());
         this.setParseTldUids(other.getParseTldUids());
-        this.setSequentialScheduler(other.getSequentialScheduler());
         this.setCollectTimingDetails(other.getCollectTimingDetails());
         this.setLogTimingDetails(other.getLogTimingDetails());
         this.setSendTimingToStatsd(other.getSendTimingToStatsd());
@@ -2274,14 +2272,6 @@ public void setReduceTypeMetadataPerShard(boolean reduceTypeMetadataPerShard) {
         this.reduceTypeMetadataPerShard = reduceTypeMetadataPerShard;
     }
 
-    public boolean getSequentialScheduler() {
-        return sequentialScheduler;
-    }
-
-    public void setSequentialScheduler(boolean sequentialScheduler) {
-        this.sequentialScheduler = sequentialScheduler;
-    }
-
     public boolean getLimitAnyFieldLookups() {
         return limitAnyFieldLookups;
     }
@@ -2828,7 +2818,6 @@ public boolean equals(Object o) {
                 getReduceTypeMetadataPerShard() == that.getReduceTypeMetadataPerShard() &&
                 isRebuildDatatypeFilter() == that.isRebuildDatatypeFilter() &&
                 isRebuildDatatypeFilterPerShard() == that.isRebuildDatatypeFilterPerShard() &&
-                getSequentialScheduler() == that.getSequentialScheduler() &&
                 getCollectTimingDetails() == that.getCollectTimingDetails() &&
                 getLogTimingDetails() == that.getLogTimingDetails() &&
                 getSendTimingToStatsd() == that.getSendTimingToStatsd() &&
@@ -3036,7 +3025,6 @@ public int hashCode() {
                 getReduceTypeMetadataPerShard(),
                 isRebuildDatatypeFilter(),
                 isRebuildDatatypeFilterPerShard(),
-                getSequentialScheduler(),
                 getCollectTimingDetails(),
                 getLogTimingDetails(),
                 getSendTimingToStatsd(),

warehouse/query-core/src/main/java/datawave/query/planner/DefaultQueryPlanner.java

@@ -2513,29 +2513,61 @@ public void configureTypeMappings(ShardQueryConfiguration config, IteratorSettin
             String nonIndexedTypes = QueryOptions.buildFieldNormalizerString(nonIndexedQueryFieldsDatatypes);
             String requiredAuthsString = metadataHelper.getUsersMetadataAuthorizationSubset();
 
-            TypeMetadata typeMetadata = getTypeMetadata();
-
-            if (config.getReduceTypeMetadata() && !isPreload) {
-                Set<String> fieldsToRetain = ReduceFields.getQueryFields(config.getQueryTree());
-                typeMetadata = typeMetadata.reduce(fieldsToRetain);
-            }
-
-            String serializedTypeMetadata = typeMetadata.toString();
-
             if (compressMappings) {
                 nonIndexedTypes = QueryOptions.compressOption(nonIndexedTypes, QueryOptions.UTF8);
                 requiredAuthsString = QueryOptions.compressOption(requiredAuthsString, QueryOptions.UTF8);
-                if (!config.getReduceTypeMetadataPerShard()) {
-                    // if we're reducing later, don't compress the type metadata
-                    serializedTypeMetadata = QueryOptions.compressOption(serializedTypeMetadata, QueryOptions.UTF8);
-                }
             }
 
             addOption(cfg, QueryOptions.NON_INDEXED_DATATYPES, nonIndexedTypes, false);
-            addOption(cfg, QueryOptions.TYPE_METADATA, serializedTypeMetadata, false);
             addOption(cfg, QueryOptions.TYPE_METADATA_AUTHS, requiredAuthsString, false);
             addOption(cfg, QueryOptions.METADATA_TABLE_NAME, config.getMetadataTableName(), false);
 
+            // now handle TypeMetadata
+            boolean canReduceTypeMetadata = !config.getProjectFields().isEmpty() || !config.getDisallowlistedFields().isEmpty();
+            if (!canReduceTypeMetadata) {
+                config.setReduceTypeMetadata(false);
+                config.setReduceTypeMetadataPerShard(false);
+            }
+
+            if (!isPreload) {
+                // TypeMetadata is serialized at the end of query planning for two reasons
+                // First, the metadata is fetched in an async thread so don't wait on that during a preload
+                // Second, the query model application updates the projection fields, so that needs to happen first
+                TypeMetadata typeMetadata = getTypeMetadata();
+
+                if (canReduceTypeMetadata && (config.getReduceTypeMetadata() || config.getReduceTypeMetadataPerShard())) {
+                    // If per-shard reduction is enabled we still attempt a first-pass reduction here. This reduces
+                    // the amount of future work done by the VisitorFunction and the raw bytes passed around.
+
+                    Set<String> fieldsToRetain = new HashSet<>();
+                    if (!config.getProjectFields().isEmpty()) {
+                        // sum query fields, projection fields, and composite fields
+                        fieldsToRetain.addAll(ReduceFields.getQueryFields(config.getQueryTree()));
+                        fieldsToRetain.addAll(config.getProjectFields());
+                        fieldsToRetain.addAll(config.getCompositeToFieldMap().keySet());
+                        // GroupBy fields already added to projection at this point in planning
+                        // Unique and Excerpt fields do not affect returned fields
+                    } else {
+                        // sum all fields, remove exclude fields
+                        fieldsToRetain.addAll(typeMetadata.keySet()); // metadata fetch filtered by datatype
+                        // might need to add composite fields here
+                        fieldsToRetain.removeAll(config.getDisallowlistedFields());
+                        // GroupBy fields already added to projection at this point in planning
+                        // Unique and Excerpt fields do not affect returned fields
+                    }
+
+                    typeMetadata = typeMetadata.reduce(fieldsToRetain);
+                }
+
+                // only compress if enabled AND not reducing per shard
+                // type metadata will be serialized in the VisitorFunction
+                String serializedTypeMetadata = typeMetadata.toString();
+                if (compressMappings && !config.getReduceTypeMetadataPerShard()) {
+                    serializedTypeMetadata = QueryOptions.compressOption(serializedTypeMetadata, QueryOptions.UTF8);
+                }
+
+                addOption(cfg, QueryOptions.TYPE_METADATA, serializedTypeMetadata, false);
+            }
         } catch (IOException e) {
             QueryException qe = new QueryException(DatawaveErrorCode.TYPE_MAPPING_CONFIG_ERROR, e);
             throw new DatawaveQueryException(qe);