You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/10/16 14:12:06 UTC
(iotdb) 02/02: fix last
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch last_first_table_model
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0e38315b711dbd90b9872890d7d71977773ea36e
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Oct 16 22:11:48 2024 +0800
fix last
---
.../TableAggregationTableScanOperator.java | 105 +++++++++++++++------
.../relational/aggregation/AvgAccumulator.java | 10 +-
.../relational/aggregation/CountAccumulator.java | 6 +-
.../relational/aggregation/LastAccumulator.java | 18 ++--
.../relational/aggregation/LastByAccumulator.java | 2 +-
.../relational/aggregation/SumAccumulator.java | 8 +-
.../relational/aggregation/TableAccumulator.java | 2 +-
.../relational/aggregation/TableAggregator.java | 8 +-
.../plan/planner/TableOperatorGenerator.java | 102 ++++++++++----------
9 files changed, 156 insertions(+), 105 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
index 51d2272cef1..734c075f9ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
@@ -43,9 +43,11 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.file.metadata.statistics.StringStatistics;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
import org.apache.tsfile.read.common.block.column.LongColumn;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
@@ -101,7 +103,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
private final int maxTsBlockLineNum;
// for different aggregations aiming to same column, use this variable to point to same column
- private final int[] layoutArray;
+ private final List<Integer> aggArguments;
private QueryDataSource queryDataSource;
@@ -129,7 +131,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
GroupByTimeParameter groupByTimeParameter,
long maxReturnSize,
boolean canUseStatistics,
- int[] layoutArray) {
+ List<Integer> aggArguments) {
super(
sourceId,
@@ -161,7 +163,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
this.measurementColumnTSDataTypes =
measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
this.currentDeviceIndex = 0;
- this.layoutArray = layoutArray;
+ this.aggArguments = aggArguments;
this.timeIterator = tableTimeRangeIterator;
if (tableTimeRangeIterator.getType()
== ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR) {
@@ -400,29 +402,32 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
TsBlock inputRegion = inputTsBlock.getRegion(0, lastIndexToProcess + 1);
// TODO(beyyes) add optimization: if only agg measurement columns, no need to transfer
- Column[] valueColumns = new Column[tableAggregators.size()];
- for (int i = 0; i < tableAggregators.size(); i++) {
- ColumnSchema columnSchema = columnSchemas.get(layoutArray[i]);
+ Column[] valueColumns = new Column[aggArguments.size()];
+ for (int i = 0; i < aggArguments.size(); i++) {
+ ColumnSchema columnSchema = columnSchemas.get(aggArguments.get(i));
if (Streams.of(
TsTableColumnCategory.ID, TsTableColumnCategory.ATTRIBUTE, TsTableColumnCategory.TIME)
.anyMatch(columnSchema.getColumnCategory()::equals)) {
- valueColumns[i] = inputRegion.getTimeColumn();
+ if (isNullIdOrAttribute(i)) {
+ valueColumns[i] =
+ new RunLengthEncodedColumn(
+ new BinaryColumn(1, Optional.of(new boolean[] {true}), new Binary[] {null}),
+ inputRegion.getTimeColumn().getPositionCount());
+ } else {
+ valueColumns[i] = inputRegion.getTimeColumn();
+ }
} else {
- valueColumns[i] = inputRegion.getColumn(columnsIndexArray[i]);
+ valueColumns[i] = inputRegion.getColumn(columnsIndexArray[aggArguments.get(i)]);
}
}
+
TsBlock tsBlock =
new TsBlock(
inputRegion.getPositionCount(),
new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, inputRegion.getPositionCount()),
valueColumns);
- for (int i = 0; i < aggregators.size(); i++) {
- if (isNullIdOrAttribute(i)) {
- continue;
- }
-
- TableAggregator aggregator = aggregators.get(i);
+ for (TableAggregator aggregator : tableAggregators) {
// current agg method has been calculated
if (aggregator.hasFinalResult()) {
continue;
@@ -441,34 +446,80 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
/** ID or Attribute is null, skip this aggregation logic */
private boolean isNullIdOrAttribute(int i) {
- if (TsTableColumnCategory.ID == columnSchemas.get(layoutArray[i]).getColumnCategory()
- && deviceEntries.get(currentDeviceIndex).getNthSegment(columnsIndexArray[i] + 1) == null) {
+ if (TsTableColumnCategory.ID == columnSchemas.get(aggArguments.get(i)).getColumnCategory()
+ && deviceEntries
+ .get(currentDeviceIndex)
+ .getNthSegment(columnsIndexArray[aggArguments.get(i)] + 1)
+ == null) {
return true;
}
- return TsTableColumnCategory.ATTRIBUTE == columnSchemas.get(layoutArray[i]).getColumnCategory()
+ return TsTableColumnCategory.ATTRIBUTE
+ == columnSchemas.get(aggArguments.get(i)).getColumnCategory()
&& deviceEntries
.get(currentDeviceIndex)
.getAttributeColumnValues()
- .get(columnsIndexArray[i])
+ .get(columnsIndexArray[aggArguments.get(i)])
== null;
}
protected void calcFromStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
- for (int i = 0; i < tableAggregators.size(); i++) {
- if (isNullIdOrAttribute(i)) {
- continue;
- }
+ int idx = -1;
+
+ for (TableAggregator aggregator : tableAggregators) {
- TableAggregator aggregator = tableAggregators.get(i);
if (aggregator.hasFinalResult()) {
continue;
}
- aggregator.processStatistics(
- columnSchemas.get(layoutArray[i]).getColumnCategory() == TsTableColumnCategory.MEASUREMENT
- ? valueStatistics[columnsIndexArray[i]]
- : timeStatistics);
+ Statistics[] statisticsArray = new Statistics[aggregator.getChannelCount()];
+
+ for (int i = 0; i < aggregator.getChannelCount(); i++) {
+ idx++;
+
+ TsTableColumnCategory columnSchemaCategory =
+ columnSchemas.get(aggArguments.get(idx)).getColumnCategory();
+ if (columnSchemaCategory == TsTableColumnCategory.TIME) {
+ statisticsArray[i] = timeStatistics;
+ } else if (columnSchemaCategory == TsTableColumnCategory.ATTRIBUTE) {
+ String attr =
+ deviceEntries
+ .get(currentDeviceIndex)
+ .getAttributeColumnValues()
+ .get(columnsIndexArray[aggArguments.get(i)]);
+ if (attr == null) {
+ statisticsArray[i] = null;
+ } else {
+ StringStatistics stringStatics = new StringStatistics();
+ stringStatics.setCount((int) timeStatistics.getCount());
+ Binary v = new Binary(attr.getBytes());
+ stringStatics.initializeStats(v, v, v, v);
+ statisticsArray[i] = stringStatics;
+ }
+ } else if (columnSchemaCategory == TsTableColumnCategory.ID) {
+ // TODO avoid create deviceStatics multi times
+ String id =
+ (String)
+ deviceEntries
+ .get(currentDeviceIndex)
+ .getNthSegment(columnsIndexArray[aggArguments.get(idx)] + 1);
+ if (id == null) {
+ statisticsArray[i] = null;
+ } else {
+ StringStatistics stringStatics = new StringStatistics();
+ stringStatics.setCount((int) timeStatistics.getCount());
+ stringStatics.setStartTime(timeStatistics.getStartTime());
+ stringStatics.setEndTime(timeStatistics.getEndTime());
+ Binary v = new Binary(id.getBytes());
+ stringStatics.initializeStats(v, v, v, v);
+ statisticsArray[i] = stringStatics;
+ }
+ } else {
+ statisticsArray[i] = valueStatistics[columnsIndexArray[aggArguments.get(idx)]];
+ }
+ }
+
+ aggregator.processStatistics(statisticsArray);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
index 283fb598b69..4c33e6ac371 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
@@ -137,16 +137,16 @@ public class AvgAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
+ public void addStatistics(Statistics[] statistics) {
if (statistics == null) {
return;
}
initResult = true;
- countValue += statistics.getCount();
- if (statistics instanceof IntegerStatistics) {
- sumValue += statistics.getSumLongValue();
+ countValue += statistics[0].getCount();
+ if (statistics[0] instanceof IntegerStatistics) {
+ sumValue += statistics[0].getSumLongValue();
} else {
- sumValue += statistics.getSumDoubleValue();
+ sumValue += statistics[0].getSumDoubleValue();
}
if (countValue == 0) {
initResult = false;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
index b179e76a38c..0d1c8de9e47 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
@@ -81,11 +81,11 @@ public class CountAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
- if (statistics == null) {
+ public void addStatistics(Statistics[] statistics) {
+ if (statistics[0] == null) {
return;
}
- countState += statistics.getCount();
+ countState += statistics[0].getCount();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
index 7bcb4d45f4f..a2100c899d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
@@ -197,32 +197,34 @@ public class LastAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
- if (statistics == null) {
+ public void addStatistics(Statistics[] statistics) {
+ if (statistics[0] == null) {
return;
}
switch (seriesDataType) {
case INT32:
case DATE:
- updateIntLastValue((int) statistics.getLastValue(), statistics.getEndTime());
+ updateIntLastValue((int) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case INT64:
+ updateLongLastValue((long) statistics[0].getLastValue(), statistics[0].getEndTime());
+ break;
case TIMESTAMP:
- updateLongLastValue((long) statistics.getLastValue(), statistics.getEndTime());
+ updateLongLastValue(statistics[0].getEndTime(), statistics[0].getEndTime());
break;
case FLOAT:
- updateFloatLastValue((float) statistics.getLastValue(), statistics.getEndTime());
+ updateFloatLastValue((float) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case DOUBLE:
- updateDoubleLastValue((double) statistics.getLastValue(), statistics.getEndTime());
+ updateDoubleLastValue((double) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case TEXT:
case BLOB:
case STRING:
- updateBinaryLastValue((Binary) statistics.getLastValue(), statistics.getEndTime());
+ updateBinaryLastValue((Binary) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case BOOLEAN:
- updateBooleanLastValue((boolean) statistics.getLastValue(), statistics.getEndTime());
+ updateBooleanLastValue((boolean) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
default:
throw new UnSupportedDataTypeException(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
index c594eca1c74..dd475864574 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
@@ -53,7 +53,7 @@ public class LastByAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {}
+ public void addStatistics(Statistics[] statistics) {}
@Override
public void reset() {}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
index e76ad72d551..ad89c40c5f1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
@@ -113,15 +113,15 @@ public class SumAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
+ public void addStatistics(Statistics[] statistics) {
if (statistics == null) {
return;
}
initResult = true;
- if (statistics instanceof IntegerStatistics) {
- sumValue += statistics.getSumLongValue();
+ if (statistics[0] instanceof IntegerStatistics) {
+ sumValue += statistics[0].getSumLongValue();
} else {
- sumValue += statistics.getSumDoubleValue();
+ sumValue += statistics[0].getSumDoubleValue();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
index 30cb908b43b..38c14de541a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
@@ -40,7 +40,7 @@ public interface TableAccumulator {
* This method can only be used in AggTableScan, it will use different statistics based on the
* type of Accumulator.
*/
- void addStatistics(Statistics statistics);
+ void addStatistics(Statistics[] statistics);
void reset();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
index fae3dee8de0..d9cba3bbbb6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
@@ -81,9 +81,7 @@ public class TableAggregator {
}
}
- /** Used for AggregateTableScanOperator. */
- public void processStatistics(Statistics statistics) {
- checkArgument(inputChannels.length == 1, "expected 1 input channel for processStatistics");
+ public void processStatistics(Statistics[] statistics) {
accumulator.addStatistics(statistics);
}
@@ -98,4 +96,8 @@ public class TableAggregator {
public long getEstimatedSize() {
return accumulator.getEstimatedSize();
}
+
+ public int getChannelCount() {
+ return this.inputChannels.length;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 86f89288fb0..8a0c5176a71 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -1391,7 +1391,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Map<Symbol, Integer> columnLayout = new HashMap<>(node.getAggregations().size());
int aggregationsCount = node.getAggregations().size();
- int[] layoutArray = new int[aggregationsCount];
+ List<Integer> aggColumnIndexes = new ArrayList<>();
int channel = 0;
int idx = -1;
int measurementColumnCount = 0;
@@ -1399,64 +1399,59 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Map<Symbol, Integer> idAndAttributeColumnsIndexMap = node.getIdAndAttributeIndexMap();
Map<Symbol, ColumnSchema> columnSchemaMap = node.getAssignments();
List<ColumnSchema> columnSchemas = new ArrayList<>(aggregationsCount);
- int[] columnsIndexArray = new int[aggregationsCount];
+ int[] columnsIndexArray = new int[aggregationsCount * 2];
List<String> measurementColumnNames = new ArrayList<>();
List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
- // List<Integer>
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
- idx++;
- Symbol symbol = Symbol.from(entry.getValue().getArguments().get(0));
- ColumnSchema schema = requireNonNull(columnSchemaMap.get(symbol), symbol + " is null");
-
String funcName = entry.getValue().getResolvedFunction().getSignature().getName();
- if (schema.getType().equals(BlobType.BLOB)
- && (FIRST.equals(funcName) || LAST.equals(funcName))) {
- // first/last aggregation with BLOB type can not use statistics
- canUseStatistic = false;
- }
- switch (schema.getColumnCategory()) {
- case ID:
- case ATTRIBUTE:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = columnsIndexArray[columnLayout.get(symbol)];
- } else {
- columnsIndexArray[idx] =
- requireNonNull(idAndAttributeColumnsIndexMap.get(symbol), symbol + " is null");
- columnSchemas.add(schema);
- }
- break;
- case MEASUREMENT:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = columnsIndexArray[columnLayout.get(symbol)];
- } else {
- columnsIndexArray[idx] = measurementColumnCount;
- measurementColumnCount++;
- measurementColumnNames.add(symbol.getName());
- measurementSchemas.add(
- new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
- columnSchemas.add(schema);
- }
- break;
- case TIME:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = -1;
- } else {
- columnsIndexArray[idx] = -1;
- columnSchemas.add(schema);
- }
- break;
- default:
- throw new IllegalArgumentException(
- "Unexpected column category: " + schema.getColumnCategory());
- }
+ for (Expression argument : entry.getValue().getArguments()) {
+ idx++;
+ Symbol symbol = Symbol.from(argument);
+ ColumnSchema schema = requireNonNull(columnSchemaMap.get(symbol), symbol + " is null");
+ if (schema.getType().equals(BlobType.BLOB)
+ && (FIRST.equals(funcName) || LAST.equals(funcName))) {
+ // first/last aggregation with BLOB type can not use statistics
+ canUseStatistic = false;
+ }
- if (!columnLayout.containsKey(symbol)) {
- layoutArray[idx] = channel;
- columnLayout.put(symbol, channel++);
- } else {
- layoutArray[idx] = columnLayout.get(symbol);
+ switch (schema.getColumnCategory()) {
+ case ID:
+ case ATTRIBUTE:
+ if (!columnLayout.containsKey(symbol)) {
+ columnsIndexArray[channel] =
+ requireNonNull(idAndAttributeColumnsIndexMap.get(symbol), symbol + " is null");
+ columnSchemas.add(schema);
+ }
+ break;
+ case MEASUREMENT:
+ if (!columnLayout.containsKey(symbol)) {
+ columnsIndexArray[channel] = measurementColumnCount;
+ measurementColumnCount++;
+ measurementColumnNames.add(symbol.getName());
+ measurementSchemas.add(
+ new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ columnSchemas.add(schema);
+ }
+ break;
+ case TIME:
+ if (!columnLayout.containsKey(symbol)) {
+ columnsIndexArray[channel] = -1;
+ columnSchemas.add(schema);
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unexpected column category: " + schema.getColumnCategory());
+ }
+
+ if (!columnLayout.containsKey(symbol)) {
+ aggColumnIndexes.add(channel);
+ columnLayout.put(symbol, channel++);
+ } else {
+ aggColumnIndexes.add(columnLayout.get(symbol));
+ }
}
aggregators.add(
@@ -1468,6 +1463,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
context.getTypeProvider()));
}
+ // TODO if this needed?
for (Map.Entry<Symbol, ColumnSchema> entry : node.getAssignments().entrySet()) {
if (!columnLayout.containsKey(entry.getKey())
&& entry.getValue().getColumnCategory() == MEASUREMENT) {
@@ -1566,7 +1562,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
null,
calculateMaxAggregationResultSize(),
canUseStatistic,
- layoutArray);
+ aggColumnIndexes);
((DataDriverContext) context.getDriverContext()).addSourceOperator(aggTableScanOperator);