This is an automated email from the ASF dual-hosted git repository. hui pushed a commit to branch lmh/FixIntoMem in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3f0e8c19a603b2d1b177cbc5ecec023386edd57c Author: Minghui Liu <[email protected]> AuthorDate: Fri Apr 7 10:45:57 2023 +0800 opt memory estimate of into operator --- .../operator/process/AbstractIntoOperator.java | 58 +++++++++++++++------- .../operator/process/DeviceViewIntoOperator.java | 7 +-- .../execution/operator/process/IntoOperator.java | 7 +-- .../db/mpp/plan/planner/OperatorTreeGenerator.java | 14 ++---- 4 files changed, 50 insertions(+), 36 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractIntoOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractIntoOperator.java index 3d65c1659c..7a439daa3c 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractIntoOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractIntoOperator.java @@ -78,8 +78,9 @@ public abstract class AbstractIntoOperator implements ProcessOperator { protected boolean finished = false; - private final long maxRetainedSize; - private final long maxReturnSize; + protected int maxRowNumberInStatement; + private long maxRetainedSize; + private long maxReturnSize; protected final List<Type> typeConvertors; @@ -89,7 +90,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { List<TSDataType> inputColumnTypes, Map<String, InputLocation> sourceColumnToInputLocationMap, ExecutorService intoOperationExecutor, - long maxStatementSize) { + long statementSizePerLine) { this.operatorContext = operatorContext; this.child = child; this.typeConvertors = @@ -97,7 +98,23 @@ public abstract class AbstractIntoOperator implements ProcessOperator { this.sourceColumnToInputLocationMap = sourceColumnToInputLocationMap; this.writeOperationExecutor = intoOperationExecutor; + initMemoryEstimates(statementSizePerLine); + } + + private void initMemoryEstimates(long statementSizePerLine) { + long intoOperationBufferSizeInByte = + IoTDBDescriptor.getInstance().getConfig().getIntoOperationBufferSizeInByte(); + long memAllowedMaxRowNumber = intoOperationBufferSizeInByte / statementSizePerLine; + if (memAllowedMaxRowNumber > Integer.MAX_VALUE) { + memAllowedMaxRowNumber = Integer.MAX_VALUE; + } + int maxRowNumberInStatement = + Math.min( + (int) memAllowedMaxRowNumber, + IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit()); + long maxStatementSize = maxRowNumberInStatement * statementSizePerLine; + this.maxRowNumberInStatement = maxRowNumberInStatement; this.maxRetainedSize = child.calculateMaxReturnSize() + maxStatementSize; this.maxReturnSize = DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES; } @@ -209,7 +226,8 @@ public abstract class AbstractIntoOperator implements ProcessOperator { Map<PartialPath, Map<String, InputLocation>> targetPathToSourceInputLocationMap, Map<PartialPath, Map<String, TSDataType>> targetPathToDataTypeMap, Map<String, Boolean> targetDeviceToAlignedMap, - List<Type> sourceTypeConvertors) { + List<Type> sourceTypeConvertors, + int maxRowNumberInStatement) { List<InsertTabletStatementGenerator> insertTabletStatementGenerators = new ArrayList<>(targetPathToSourceInputLocationMap.size()); for (Map.Entry<PartialPath, Map<String, InputLocation>> entry : @@ -221,7 +239,8 @@ public abstract class AbstractIntoOperator implements ProcessOperator { entry.getValue(), targetPathToDataTypeMap.get(targetDevice), targetDeviceToAlignedMap.get(targetDevice.toString()), - sourceTypeConvertors); + sourceTypeConvertors, + maxRowNumberInStatement); insertTabletStatementGenerators.add(generator); } return insertTabletStatementGenerators; @@ -324,8 +343,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { public static class InsertTabletStatementGenerator { - private final int TABLET_ROW_LIMIT = - IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit(); + private final int rowLimit; private final PartialPath devicePath; private final boolean isAligned; @@ -348,7 +366,8 @@ public abstract class AbstractIntoOperator implements ProcessOperator { Map<String, InputLocation> measurementToInputLocationMap, Map<String, TSDataType> measurementToDataTypeMap, Boolean isAligned, - List<Type> sourceTypeConvertors) { + List<Type> sourceTypeConvertors, + int rowLimit) { this.devicePath = devicePath; this.isAligned = isAligned; this.measurements = measurementToInputLocationMap.keySet().toArray(new String[0]); @@ -359,32 +378,33 @@ public abstract class AbstractIntoOperator implements ProcessOperator { writtenCounter.put(measurement, new AtomicInteger(0)); } this.sourceTypeConvertors = sourceTypeConvertors; + this.rowLimit = rowLimit; this.reset(); } public void reset() { this.rowCount = 0; - this.times = new long[TABLET_ROW_LIMIT]; + this.times = new long[rowLimit]; this.columns = new Object[this.measurements.length]; for (int i = 0; i < this.measurements.length; i++) { switch (dataTypes[i]) { case BOOLEAN: - columns[i] = new boolean[TABLET_ROW_LIMIT]; + columns[i] = new boolean[rowLimit]; break; case INT32: - columns[i] = new int[TABLET_ROW_LIMIT]; + columns[i] = new int[rowLimit]; break; case INT64: - columns[i] = new long[TABLET_ROW_LIMIT]; + columns[i] = new long[rowLimit]; break; case FLOAT: - columns[i] = new float[TABLET_ROW_LIMIT]; + columns[i] = new float[rowLimit]; break; case DOUBLE: - columns[i] = new double[TABLET_ROW_LIMIT]; + columns[i] = new double[rowLimit]; break; case TEXT: - columns[i] = new Binary[TABLET_ROW_LIMIT]; + columns[i] = new Binary[rowLimit]; Arrays.fill((Binary[]) columns[i], Binary.EMPTY_VALUE); break; default: @@ -394,7 +414,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { } this.bitMaps = new BitMap[this.measurements.length]; for (int i = 0; i < this.bitMaps.length; ++i) { - this.bitMaps[i] = new BitMap(TABLET_ROW_LIMIT); + this.bitMaps[i] = new BitMap(rowLimit); this.bitMaps[i].markAll(); } } @@ -452,7 +472,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { ++rowCount; ++lastReadIndex; - if (rowCount == TABLET_ROW_LIMIT) { + if (rowCount == rowLimit) { break; } } @@ -460,7 +480,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { } public boolean isFull() { - return rowCount == TABLET_ROW_LIMIT; + return rowCount == rowLimit; } public boolean isEmpty() { @@ -475,7 +495,7 @@ public abstract class AbstractIntoOperator implements ProcessOperator { insertTabletStatement.setDataTypes(dataTypes); insertTabletStatement.setRowCount(rowCount); - if (rowCount != TABLET_ROW_LIMIT) { + if (rowCount != rowLimit) { times = Arrays.copyOf(times, rowCount); for (int i = 0; i < columns.length; i++) { bitMaps[i] = bitMaps[i].getRegion(0, rowCount); diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceViewIntoOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceViewIntoOperator.java index 0ddd7ccc31..5f2d202219 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceViewIntoOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceViewIntoOperator.java @@ -64,14 +64,14 @@ public class DeviceViewIntoOperator extends AbstractIntoOperator { Map<String, List<Pair<String, PartialPath>>> deviceToSourceTargetPathPairListMap, Map<String, InputLocation> sourceColumnToInputLocationMap, ExecutorService intoOperationExecutor, - long maxStatementSize) { + long statementSizePerLine) { super( operatorContext, child, inputColumnTypes, sourceColumnToInputLocationMap, intoOperationExecutor, - maxStatementSize); + statementSizePerLine); this.deviceToTargetPathSourceInputLocationMap = deviceToTargetPathSourceInputLocationMap; this.deviceToTargetPathDataTypeMap = deviceToTargetPathDataTypeMap; this.targetDeviceToAlignedMap = targetDeviceToAlignedMap; @@ -148,7 +148,8 @@ public class DeviceViewIntoOperator extends AbstractIntoOperator { targetPathToSourceInputLocationMap, targetPathToDataTypeMap, targetDeviceToAlignedMap, - typeConvertors); + typeConvertors, + maxRowNumberInStatement); } private void updateResultTsBlock() { diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/IntoOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/IntoOperator.java index 34a8edc88a..d5e65d262e 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/IntoOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/IntoOperator.java @@ -52,21 +52,22 @@ public class IntoOperator extends AbstractIntoOperator { List<Pair<String, PartialPath>> sourceTargetPathPairList, Map<String, InputLocation> sourceColumnToInputLocationMap, ExecutorService intoOperationExecutor, - long maxStatementSize) { + long statementSizePerLine) { super( operatorContext, child, inputColumnTypes, sourceColumnToInputLocationMap, intoOperationExecutor, - maxStatementSize); + statementSizePerLine); this.sourceTargetPathPairList = sourceTargetPathPairList; insertTabletStatementGenerators = constructInsertTabletStatementGenerators( targetPathToSourceInputLocationMap, targetPathToDataTypeMap, targetDeviceToAlignedMap, - typeConvertors); + typeConvertors, + maxRowNumberInStatement); } @Override diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java index 1083fafcfc..43c9c81454 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java @@ -22,7 +22,6 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; -import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache; import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory; @@ -1639,10 +1638,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP Map<PartialPath, Map<String, TSDataType>> targetPathToDataTypeMap = intoPathDescriptor.getTargetPathToDataTypeMap(); - - int rowLimit = - IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit(); - long maxStatementSize = calculateStatementSizePerLine(targetPathToDataTypeMap) * rowLimit; + long statementSizePerLine = calculateStatementSizePerLine(targetPathToDataTypeMap); context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1); @@ -1656,7 +1652,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP intoPathDescriptor.getSourceTargetPathPairList(), sourceColumnToInputLocationMap, FragmentInstanceManager.getInstance().getIntoOperationExecutor(), - maxStatementSize); + statementSizePerLine); } @Override @@ -1697,10 +1693,6 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP calculateStatementSizePerLine(deviceToTargetPathDataTypeMap.get(sourceDevice)); } - int rowLimit = - IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit(); - long maxStatementSize = statementSizePerLine * rowLimit; - context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1); return new DeviceViewIntoOperator( operatorContext, @@ -1712,7 +1704,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP deviceViewIntoPathDescriptor.getDeviceToSourceTargetPathPairListMap(), sourceColumnToInputLocationMap, FragmentInstanceManager.getInstance().getIntoOperationExecutor(), - maxStatementSize); + statementSizePerLine); } private Map<String, InputLocation> constructSourceColumnToInputLocationMap(PlanNode node) {
