VGalaxies commented on code in PR #17238: URL: https://github.com/apache/iotdb/pull/17238#discussion_r3346562381
########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java: ########## @@ -0,0 +1,606 @@ +/* + * 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.iotdb.db.subscription.broker.consensus; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.regex.Pattern; + +/** Converts IoTConsensus WAL log entries (InsertNode) to Tablet format for subscription. */ +public class ConsensusLogToTabletConverter { + + private static final Logger LOGGER = LoggerFactory.getLogger(ConsensusLogToTabletConverter.class); + + private final TreePattern treePattern; + private final TablePattern tablePattern; + private final Pattern tableColumnPattern; + + /** + * The actual database name of the DataRegion this converter processes (table-model format without + * "root." prefix). Null for tree-model topics. + */ + private final String databaseName; + + public ConsensusLogToTabletConverter( + final TreePattern treePattern, + final TablePattern tablePattern, + final Pattern tableColumnPattern, + final String databaseName) { + this.treePattern = treePattern; + this.tablePattern = tablePattern; + this.tableColumnPattern = tableColumnPattern; + this.databaseName = databaseName; + } + + public String getDatabaseName() { + return databaseName; + } + + static String safeDeviceIdForLog(final InsertNode node) { + try { + final Object deviceId = node.getDeviceID(); + return deviceId != null ? deviceId.toString() : "null"; + } catch (final Exception e) { + return "N/A(" + node.getType() + ")"; + } + } + + public List<Tablet> convert(final InsertNode insertNode) { + if (Objects.isNull(insertNode)) { + return Collections.emptyList(); + } + + final PlanNodeType nodeType = insertNode.getType(); + if (nodeType == null) { + LOGGER.warn("InsertNode type is null, skipping conversion"); + return Collections.emptyList(); + } + + LOGGER.debug( + "ConsensusLogToTabletConverter: converting InsertNode type={}, deviceId={}", + nodeType, + safeDeviceIdForLog(insertNode)); + + switch (nodeType) { + case INSERT_ROW: + return convertInsertRowNode((InsertRowNode) insertNode); + case INSERT_TABLET: + return convertInsertTabletNode((InsertTabletNode) insertNode); + case INSERT_ROWS: + return convertInsertRowsNode((InsertRowsNode) insertNode); + case INSERT_ROWS_OF_ONE_DEVICE: + return convertInsertRowsOfOneDeviceNode((InsertRowsOfOneDeviceNode) insertNode); + case INSERT_MULTI_TABLET: + return convertInsertMultiTabletsNode((InsertMultiTabletsNode) insertNode); + case RELATIONAL_INSERT_ROW: + return convertRelationalInsertRowNode((RelationalInsertRowNode) insertNode); + case RELATIONAL_INSERT_TABLET: + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) insertNode); + case RELATIONAL_INSERT_ROWS: + return convertRelationalInsertRowsNode((RelationalInsertRowsNode) insertNode); + default: + LOGGER.debug("Unsupported InsertNode type for subscription: {}", nodeType); + return Collections.emptyList(); + } + } + + // ======================== Tree Model Conversion ======================== + + private List<Tablet> convertInsertRowNode(final InsertRowNode node) { + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final long time = node.getTime(); + + // Determine which columns match the pattern + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final Object[] values = node.getValues(); + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + // Build Tablet with matched columns + final int columnCount = matchedColumnIndices.size(); + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + final Tablet tablet = new Tablet(deviceId.toString(), schemas, 1 /* maxRowNumber */); + tablet.addTimestamp(0, time); + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = matchedColumnIndices.get(i); + final Object value = values[originalColIdx]; + if (value == null) { + if (tablet.getBitMaps() == null) { + tablet.initBitMaps(); + } + tablet.getBitMaps()[i].mark(0); + } else { + addValueToTablet(tablet, 0, i, dataTypes[originalColIdx], value); + } + } + tablet.setRowSize(1); + + return Collections.singletonList(tablet); + } + + private List<Tablet> convertInsertTabletNode(final InsertTabletNode node) { + if (node instanceof RelationalInsertTabletNode) { + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) node); + } + + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final long[] times = node.getTimes(); + final Object[] columns = node.getColumns(); + final BitMap[] bitMaps = node.getBitMaps(); + final int rowCount = node.getRowCount(); + + // Column filtering + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + final int columnCount = matchedColumnIndices.size(); + final boolean allColumnsMatch = (columnCount == measurements.length); + + // Build schemas (always needed) + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + // Build column arrays and bitmaps using bulk copy + final long[] newTimes = Arrays.copyOf(times, rowCount); + final Object[] newColumns = new Object[columnCount]; + final BitMap[] newBitMaps = new BitMap[columnCount]; + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = allColumnsMatch ? i : matchedColumnIndices.get(i); + newColumns[i] = copyColumnArray(dataTypes[originalColIdx], columns[originalColIdx], rowCount); + if (bitMaps != null && bitMaps[originalColIdx] != null) { + newBitMaps[i] = new BitMap(rowCount); + BitMap.copyOfRange(bitMaps[originalColIdx], 0, newBitMaps[i], 0, rowCount); + } + } Review Comment: Done in 5ea428025e. The complete-tablet path now reuses the original time/column/bitmap arrays on the server-side read path; copies are kept only when column filtering changes the tablet shape. A comment was added to document that tradeoff. ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( Review Comment: Done in 5ea428025e. Renamed PipeSubscribeHeartbeatReq to SubscriptionHeartbeatReq and updated the call sites/log wording accordingly. ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( + final List<SubscriptionCommitContext> processorBufferedCommitContexts) throws IOException { + final PipeSubscribeHeartbeatReq req = toTPipeSubscribeReq(); + req.processorBufferedCommitContexts = + Objects.nonNull(processorBufferedCommitContexts) + ? processorBufferedCommitContexts + : new ArrayList<>(); Review Comment: Done in 5ea428025e. The default buffered commit context list now uses Collections.emptyList(). ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java: ########## @@ -0,0 +1,606 @@ +/* + * 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.iotdb.db.subscription.broker.consensus; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.regex.Pattern; + +/** Converts IoTConsensus WAL log entries (InsertNode) to Tablet format for subscription. */ +public class ConsensusLogToTabletConverter { + + private static final Logger LOGGER = LoggerFactory.getLogger(ConsensusLogToTabletConverter.class); + + private final TreePattern treePattern; + private final TablePattern tablePattern; + private final Pattern tableColumnPattern; + + /** + * The actual database name of the DataRegion this converter processes (table-model format without + * "root." prefix). Null for tree-model topics. + */ + private final String databaseName; + + public ConsensusLogToTabletConverter( + final TreePattern treePattern, + final TablePattern tablePattern, + final Pattern tableColumnPattern, + final String databaseName) { + this.treePattern = treePattern; + this.tablePattern = tablePattern; + this.tableColumnPattern = tableColumnPattern; + this.databaseName = databaseName; + } + + public String getDatabaseName() { + return databaseName; + } + + static String safeDeviceIdForLog(final InsertNode node) { + try { + final Object deviceId = node.getDeviceID(); + return deviceId != null ? deviceId.toString() : "null"; + } catch (final Exception e) { + return "N/A(" + node.getType() + ")"; + } + } + + public List<Tablet> convert(final InsertNode insertNode) { + if (Objects.isNull(insertNode)) { + return Collections.emptyList(); + } + + final PlanNodeType nodeType = insertNode.getType(); + if (nodeType == null) { + LOGGER.warn("InsertNode type is null, skipping conversion"); + return Collections.emptyList(); + } + + LOGGER.debug( + "ConsensusLogToTabletConverter: converting InsertNode type={}, deviceId={}", + nodeType, + safeDeviceIdForLog(insertNode)); + + switch (nodeType) { + case INSERT_ROW: + return convertInsertRowNode((InsertRowNode) insertNode); + case INSERT_TABLET: + return convertInsertTabletNode((InsertTabletNode) insertNode); + case INSERT_ROWS: + return convertInsertRowsNode((InsertRowsNode) insertNode); + case INSERT_ROWS_OF_ONE_DEVICE: + return convertInsertRowsOfOneDeviceNode((InsertRowsOfOneDeviceNode) insertNode); + case INSERT_MULTI_TABLET: + return convertInsertMultiTabletsNode((InsertMultiTabletsNode) insertNode); + case RELATIONAL_INSERT_ROW: + return convertRelationalInsertRowNode((RelationalInsertRowNode) insertNode); + case RELATIONAL_INSERT_TABLET: + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) insertNode); + case RELATIONAL_INSERT_ROWS: + return convertRelationalInsertRowsNode((RelationalInsertRowsNode) insertNode); + default: + LOGGER.debug("Unsupported InsertNode type for subscription: {}", nodeType); + return Collections.emptyList(); + } + } + + // ======================== Tree Model Conversion ======================== + + private List<Tablet> convertInsertRowNode(final InsertRowNode node) { + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final long time = node.getTime(); + + // Determine which columns match the pattern + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final Object[] values = node.getValues(); + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + // Build Tablet with matched columns + final int columnCount = matchedColumnIndices.size(); + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + final Tablet tablet = new Tablet(deviceId.toString(), schemas, 1 /* maxRowNumber */); + tablet.addTimestamp(0, time); + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = matchedColumnIndices.get(i); + final Object value = values[originalColIdx]; + if (value == null) { + if (tablet.getBitMaps() == null) { + tablet.initBitMaps(); + } + tablet.getBitMaps()[i].mark(0); + } else { + addValueToTablet(tablet, 0, i, dataTypes[originalColIdx], value); + } + } + tablet.setRowSize(1); + + return Collections.singletonList(tablet); + } + + private List<Tablet> convertInsertTabletNode(final InsertTabletNode node) { + if (node instanceof RelationalInsertTabletNode) { + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) node); + } + + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final long[] times = node.getTimes(); + final Object[] columns = node.getColumns(); + final BitMap[] bitMaps = node.getBitMaps(); + final int rowCount = node.getRowCount(); + + // Column filtering + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + final int columnCount = matchedColumnIndices.size(); + final boolean allColumnsMatch = (columnCount == measurements.length); + + // Build schemas (always needed) + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + // Build column arrays and bitmaps using bulk copy + final long[] newTimes = Arrays.copyOf(times, rowCount); + final Object[] newColumns = new Object[columnCount]; + final BitMap[] newBitMaps = new BitMap[columnCount]; + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = allColumnsMatch ? i : matchedColumnIndices.get(i); + newColumns[i] = copyColumnArray(dataTypes[originalColIdx], columns[originalColIdx], rowCount); + if (bitMaps != null && bitMaps[originalColIdx] != null) { + newBitMaps[i] = new BitMap(rowCount); + BitMap.copyOfRange(bitMaps[originalColIdx], 0, newBitMaps[i], 0, rowCount); + } + } Review Comment: Done in 5ea428025e. The complete-tablet path now reuses the original time/column/bitmap arrays on the server-side read path; copies are kept only when column filtering changes the tablet shape. A comment was added to document that tradeoff. ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( Review Comment: Done in 5ea428025e. Renamed PipeSubscribeHeartbeatReq to SubscriptionHeartbeatReq and updated the call sites/log wording accordingly. ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( + final List<SubscriptionCommitContext> processorBufferedCommitContexts) throws IOException { + final PipeSubscribeHeartbeatReq req = toTPipeSubscribeReq(); + req.processorBufferedCommitContexts = + Objects.nonNull(processorBufferedCommitContexts) + ? processorBufferedCommitContexts + : new ArrayList<>(); Review Comment: Done in 5ea428025e. The default buffered commit context list now uses Collections.emptyList(). ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( Review Comment: Done in 5ea428025e. Renamed PipeSubscribeHeartbeatReq to SubscriptionHeartbeatReq and updated the call sites/log wording accordingly. ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java: ########## @@ -0,0 +1,606 @@ +/* + * 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.iotdb.db.subscription.broker.consensus; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.regex.Pattern; + +/** Converts IoTConsensus WAL log entries (InsertNode) to Tablet format for subscription. */ +public class ConsensusLogToTabletConverter { + + private static final Logger LOGGER = LoggerFactory.getLogger(ConsensusLogToTabletConverter.class); + + private final TreePattern treePattern; + private final TablePattern tablePattern; + private final Pattern tableColumnPattern; + + /** + * The actual database name of the DataRegion this converter processes (table-model format without + * "root." prefix). Null for tree-model topics. + */ + private final String databaseName; + + public ConsensusLogToTabletConverter( + final TreePattern treePattern, + final TablePattern tablePattern, + final Pattern tableColumnPattern, + final String databaseName) { + this.treePattern = treePattern; + this.tablePattern = tablePattern; + this.tableColumnPattern = tableColumnPattern; + this.databaseName = databaseName; + } + + public String getDatabaseName() { + return databaseName; + } + + static String safeDeviceIdForLog(final InsertNode node) { + try { + final Object deviceId = node.getDeviceID(); + return deviceId != null ? deviceId.toString() : "null"; + } catch (final Exception e) { + return "N/A(" + node.getType() + ")"; + } + } + + public List<Tablet> convert(final InsertNode insertNode) { + if (Objects.isNull(insertNode)) { + return Collections.emptyList(); + } + + final PlanNodeType nodeType = insertNode.getType(); + if (nodeType == null) { + LOGGER.warn("InsertNode type is null, skipping conversion"); + return Collections.emptyList(); + } + + LOGGER.debug( + "ConsensusLogToTabletConverter: converting InsertNode type={}, deviceId={}", + nodeType, + safeDeviceIdForLog(insertNode)); + + switch (nodeType) { + case INSERT_ROW: + return convertInsertRowNode((InsertRowNode) insertNode); + case INSERT_TABLET: + return convertInsertTabletNode((InsertTabletNode) insertNode); + case INSERT_ROWS: + return convertInsertRowsNode((InsertRowsNode) insertNode); + case INSERT_ROWS_OF_ONE_DEVICE: + return convertInsertRowsOfOneDeviceNode((InsertRowsOfOneDeviceNode) insertNode); + case INSERT_MULTI_TABLET: + return convertInsertMultiTabletsNode((InsertMultiTabletsNode) insertNode); + case RELATIONAL_INSERT_ROW: + return convertRelationalInsertRowNode((RelationalInsertRowNode) insertNode); + case RELATIONAL_INSERT_TABLET: + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) insertNode); + case RELATIONAL_INSERT_ROWS: + return convertRelationalInsertRowsNode((RelationalInsertRowsNode) insertNode); + default: + LOGGER.debug("Unsupported InsertNode type for subscription: {}", nodeType); + return Collections.emptyList(); + } + } + + // ======================== Tree Model Conversion ======================== + + private List<Tablet> convertInsertRowNode(final InsertRowNode node) { + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final long time = node.getTime(); + + // Determine which columns match the pattern + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final Object[] values = node.getValues(); + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + // Build Tablet with matched columns + final int columnCount = matchedColumnIndices.size(); + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + final Tablet tablet = new Tablet(deviceId.toString(), schemas, 1 /* maxRowNumber */); + tablet.addTimestamp(0, time); + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = matchedColumnIndices.get(i); + final Object value = values[originalColIdx]; + if (value == null) { + if (tablet.getBitMaps() == null) { + tablet.initBitMaps(); + } + tablet.getBitMaps()[i].mark(0); + } else { + addValueToTablet(tablet, 0, i, dataTypes[originalColIdx], value); + } + } + tablet.setRowSize(1); + + return Collections.singletonList(tablet); + } + + private List<Tablet> convertInsertTabletNode(final InsertTabletNode node) { + if (node instanceof RelationalInsertTabletNode) { + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) node); + } + + final IDeviceID deviceId = node.getDeviceID(); + + // Device-level path filtering + if (treePattern != null && !treePattern.mayOverlapWithDevice(deviceId)) { + return Collections.emptyList(); + } + + final String[] measurements = node.getMeasurements(); + final TSDataType[] dataTypes = node.getDataTypes(); + final long[] times = node.getTimes(); + final Object[] columns = node.getColumns(); + final BitMap[] bitMaps = node.getBitMaps(); + final int rowCount = node.getRowCount(); + + // Column filtering + final List<Integer> matchedColumnIndices = getMatchedTreeColumnIndices(deviceId, measurements); + if (matchedColumnIndices.isEmpty()) { + return Collections.emptyList(); + } + + final int columnCount = matchedColumnIndices.size(); + final boolean allColumnsMatch = (columnCount == measurements.length); + + // Build schemas (always needed) + final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount); + for (final int colIdx : matchedColumnIndices) { + schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx])); + } + + // Build column arrays and bitmaps using bulk copy + final long[] newTimes = Arrays.copyOf(times, rowCount); + final Object[] newColumns = new Object[columnCount]; + final BitMap[] newBitMaps = new BitMap[columnCount]; + + for (int i = 0; i < columnCount; i++) { + final int originalColIdx = allColumnsMatch ? i : matchedColumnIndices.get(i); + newColumns[i] = copyColumnArray(dataTypes[originalColIdx], columns[originalColIdx], rowCount); + if (bitMaps != null && bitMaps[originalColIdx] != null) { + newBitMaps[i] = new BitMap(rowCount); + BitMap.copyOfRange(bitMaps[originalColIdx], 0, newBitMaps[i], 0, rowCount); + } + } Review Comment: Done in 5ea428025e. The complete-tablet path now reuses the original time/column/bitmap arrays on the server-side read path; copies are kept only when column filtering changes the tablet shape. A comment was added to document that tradeoff. ########## iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/request/PipeSubscribeHeartbeatReq.java: ########## @@ -40,11 +56,43 @@ public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq() { return req; } + /** + * Serialize the incoming parameters into `PipeSubscribeHeartbeatReq`, called by the subscription + * client. + */ + public static PipeSubscribeHeartbeatReq toTPipeSubscribeReq( + final List<SubscriptionCommitContext> processorBufferedCommitContexts) throws IOException { + final PipeSubscribeHeartbeatReq req = toTPipeSubscribeReq(); + req.processorBufferedCommitContexts = + Objects.nonNull(processorBufferedCommitContexts) + ? processorBufferedCommitContexts + : new ArrayList<>(); Review Comment: Done in 5ea428025e. The default buffered commit context list now uses Collections.emptyList(). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
