diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPreviewDataFrameTransformAction.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPreviewDataFrameTransformAction.java
index f4b93cc6ac412..dde9edb37e55c 100644
--- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPreviewDataFrameTransformAction.java
+++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPreviewDataFrameTransformAction.java
@@ -31,6 +31,7 @@
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameIndexerTransformStats;
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformConfig;
 import org.elasticsearch.xpack.core.dataframe.transforms.SourceConfig;
+import org.elasticsearch.xpack.dataframe.transforms.pivot.AggregationResultUtils;
 import org.elasticsearch.xpack.dataframe.transforms.pivot.Pivot;
 
 import java.util.List;
@@ -102,14 +103,21 @@ private void getPreview(Pivot pivot, SourceConfig source, ActionListener<List<Ma
                     pivot.buildSearchRequest(source, null, NUMBER_OF_PREVIEW_BUCKETS),
                     ActionListener.wrap(
                         r -> {
-                            final CompositeAggregation agg = r.getAggregations().get(COMPOSITE_AGGREGATION_NAME);
-                            DataFrameIndexerTransformStats stats = DataFrameIndexerTransformStats.withDefaultTransformId();
-                            // remove all internal fields
-                            List<Map<String, Object>> results = pivot.extractResults(agg, deducedMappings, stats)
-                                    .peek(record -> {
-                                        record.keySet().removeIf(k -> k.startsWith("_"));
-                                    }).collect(Collectors.toList());
-                            listener.onResponse(results);
+
+                            try {
+                                final CompositeAggregation agg = r.getAggregations().get(COMPOSITE_AGGREGATION_NAME);
+                                DataFrameIndexerTransformStats stats = DataFrameIndexerTransformStats.withDefaultTransformId();
+                                // remove all internal fields
+                                List<Map<String, Object>> results = pivot.extractResults(agg, deducedMappings, stats)
+                                        .peek(record -> {
+                                            record.keySet().removeIf(k -> k.startsWith("_"));
+                                        }).collect(Collectors.toList());
+
+                                listener.onResponse(results);
+                            } catch (AggregationResultUtils.AggregationExtractionException extractionException) {
+                                listener.onFailure(
+                                        new ElasticsearchStatusException(extractionException.getMessage(), RestStatus.BAD_REQUEST));
+                            }
                         },
                         listener::onFailure
                     ));
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/preview_transforms.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/preview_transforms.yml
index 5e58048b3bf0f..7b5c4e8cb5664 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/preview_transforms.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/preview_transforms.yml
@@ -127,3 +127,38 @@ setup:
               "aggs": {"avg_response": {"avg": {"field": "responsetime"}}}
             }
           }
+
+---
+"Test preview returns bad request with invalid agg":
+  - do:
+      catch: bad_request
+      data_frame.preview_data_frame_transform:
+        body: >
+          {
+            "source": { "index": "airline-data" },
+            "pivot": {
+              "group_by": {
+                "time": {"date_histogram": {"fixed_interval": "1h", "field": "time", "format": "yyyy-MM-DD HH"}}},
+              "aggs": {
+                "avg_response": {"avg": {"field": "responsetime"}},
+                "time.min": {"min": {"field": "time"}}
+              }
+            }
+          }
+
+  - do:
+      catch: /mixed object types of nested and non-nested fields \[time.min\]/
+      data_frame.preview_data_frame_transform:
+        body: >
+          {
+            "source": { "index": "airline-data" },
+            "pivot": {
+              "group_by": {
+                "time": {"date_histogram": {"fixed_interval": "1h", "field": "time", "format": "yyyy-MM-DD HH"}}},
+              "aggs": {
+                "avg_response": {"avg": {"field": "responsetime"}},
+                "time.min": {"min": {"field": "time"}}
+              }
+            }
+          }
+