[Bug] NPE when inserting null values for pipe data synchronization

[VGalaxies]

Search before asking

• I searched in the issues and found nothing similar.

Version

• OS version: x86_64 Linux 6.2.0-33-generic
• IoTDB version: 1.2.1

Describe the bug and provide the minimal reproduce step

1. starting two instances of IoTDB

• A datanode -> 127.0.0.1:6667
• B datanode -> 127.0.0.1:6668

2. connecting IoTDB A (6667) by cli and send

create pipe test
with connector (
    'connector'='iotdb-thrift-connector',
    'connector.ip'='127.0.0.1',
    'connector.port'='6668'
);

start pipe test;

create aligned TIMESERIES root.sg.d1(s0 float, s1 float);

insert into root.sg.d1(time, s0, s1) values (3, null, 25.34);

What did you expect to see?

no errors occur

What did you see instead?

• from IoTDB A DN (6667)

2023-09-22 15:52:35,746 [pool-37-IoTDB-Pipe-Connector-Executor-Pool-2] ERROR o.a.i.c.c.t.WrappedThreadPoolExecutor:111 - Exception in thread pool org.apache.iotdb.threadpool:type=Pipe-Connector-Executor-Pool 
org.apache.iotdb.pipe.api.exception.PipeConnectionException: PipeConnector: org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector heartbeat failed
        at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.executeOnce(PipeConnectorSubtask.java:111)
        at org.apache.iotdb.db.pipe.task.subtask.PipeSubtask.call(PipeSubtask.java:80)
        at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.call(PipeConnectorSubtask.java:82)
        at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.call(PipeConnectorSubtask.java:49)
        at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
        at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:75)
        at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.thrift.TApplicationException: Internal error processing pipeTransfer
        at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:79)
        at org.apache.iotdb.service.rpc.thrift.IClientRPCService$Client.recv_pipeTransfer(IClientRPCService.java:1665)
        at org.apache.iotdb.service.rpc.thrift.IClientRPCService$Client.pipeTransfer(IClientRPCService.java:1652)
        at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector.doTransfer(IoTDBThriftSyncConnector.java:276)
        at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector.transfer(IoTDBThriftSyncConnector.java:266)
        at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.executeOnce(PipeConnectorSubtask.java:108)
        ... 9 common frames omitted

• from IoTDB B DN (6668)

2023-09-22 15:52:35,743 [pool-31-IoTDB-ClientRPC-Processor-3] ERROR o.a.t.ProcessFunction:47 - Internal error processing pipeTransfer 
java.lang.NullPointerException: null
        at org.apache.iotdb.db.utils.CommonUtils.checkCanCastType(CommonUtils.java:113)
        at org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement.checkAndCastDataType(InsertRowStatement.java:177)
        at org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement.selfCheckDataTypes(InsertBaseStatement.java:167)
        at org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement.validateMeasurementSchema(InsertRowStatement.java:382)
        at org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaValidation.computeMeasurement(ISchemaValidation.java:34)
        at org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree.compute(ClusterSchemaTree.java:191)
        at org.apache.iotdb.db.queryengine.plan.analyze.schema.NormalSchemaFetcher.processNormalTimeSeries(NormalSchemaFetcher.java:285)
        at org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher.fetchAndComputeSchemaWithAutoCreate(ClusterSchemaFetcher.java:216)
        at org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator.validate(SchemaValidator.java:45)
        at org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.validateSchema(AnalyzeVisitor.java:2673)
        at org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.visitInsertRows(AnalyzeVisitor.java:2557)
        at org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.visitPipeEnrichedInsert(AnalyzeVisitor.java:2653)
        at org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.visitPipeEnrichedInsert(AnalyzeVisitor.java:208)
        at org.apache.iotdb.db.queryengine.plan.statement.crud.PipeEnrichedInsertBaseStatement.accept(PipeEnrichedInsertBaseStatement.java:55)
        at org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor.process(StatementVisitor.java:130)
        at org.apache.iotdb.db.queryengine.plan.analyze.Analyzer.analyze(Analyzer.java:47)
        at org.apache.iotdb.db.queryengine.plan.execution.QueryExecution.analyze(QueryExecution.java:310)
        at org.apache.iotdb.db.queryengine.plan.execution.QueryExecution.<init>(QueryExecution.java:170)
        at org.apache.iotdb.db.queryengine.plan.Coordinator.createQueryExecution(Coordinator.java:113)
        at org.apache.iotdb.db.queryengine.plan.Coordinator.execute(Coordinator.java:147)
        at org.apache.iotdb.db.pipe.receiver.thrift.IoTDBThriftReceiverV1.executeStatement(IoTDBThriftReceiverV1.java:498)
        at org.apache.iotdb.db.pipe.receiver.thrift.IoTDBThriftReceiverV1.handleTransferTabletBatch(IoTDBThriftReceiverV1.java:241)
        at org.apache.iotdb.db.pipe.receiver.thrift.IoTDBThriftReceiverV1.receive(IoTDBThriftReceiverV1.java:111)
        at org.apache.iotdb.db.pipe.receiver.thrift.IoTDBThriftReceiverAgent.receive(IoTDBThriftReceiverAgent.java:43)
        at org.apache.iotdb.db.protocol.thrift.impl.ClientRPCServiceImpl.pipeTransfer(ClientRPCServiceImpl.java:2549)
        at org.apache.iotdb.service.rpc.thrift.IClientRPCService$Processor$pipeTransfer.getResult(IClientRPCService.java:5295)
        at org.apache.iotdb.service.rpc.thrift.IClientRPCService$Processor$pipeTransfer.getResult(IClientRPCService.java:5275)
        at org.apache.thrift.ProcessFunction.process(ProcessFunction.java:38)
        at org.apache.iotdb.db.protocol.thrift.ProcessorWithMetrics.process(ProcessorWithMetrics.java:64)
        at org.apache.thrift.server.TThreadPoolServer$WorkerProcess.run(TThreadPoolServer.java:248)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[github-actions]

Hi, this is your first issue in IoTDB project. Thanks for your report. Welcome to join the community!