You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ch...@apache.org on 2025/02/05 10:06:22 UTC
(iotdb) branch udtf updated (cb66448a1d7 -> b9dd299cbe8)
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a change to branch udtf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from cb66448a1d7 add leaf operator
new 87fc892d072 save
add 5df241c044b [Table Model] Refactor UDSF/UDAF interface
add 2669666d8d2 [To dev/1.3]Script adaptation to 4 new data types (#14296) (#14630)
add f007e8a1407 Pipe: Support 'format'='tsfile' option in table model data sync sinks (#14108)
add 6db2249c12c Load: Add LoadWithConvertOnTypeMismatchIT on Table Model && Fix TSInsertTabletReq.writeToTable is not set True in PipeTransferTabletRawReq.toTPipeTransferRawReq when inserting a Tablet in Table mode (#14611)
add 487a1a21ce9 Pipe & Load: Retry when a file operation causes an exception (#14598)
add d60c3d52cb1 IoTConsensusV2: Fix adoption for table-deletion #14648
add 90cdf3bbc94 Load: Consider isGeneratedByPipe mark when executing tsfile-tablet conversion in analysis stage (#14651)
add d9214b31465 Rename interface name of FunctionArguments
add 5f8b2c08ac9 Adjust default wal_throttle_threshold_in_byte and region_migration_speed_limit_bytes_per_second (#14653)
add 815ea9c069d Implemented query for schema related Information schema & Fixed the bug that the deviceEntry get from cache does not reserve query memory
add a4ce6f7a1a3 Modify the condition to skip compaction schdule after insertion compaction task selection (#14644)
add c087c24435c Do not add the source path of a view when it can be matched by the deletion pattern (#14658)
add c60520aa41a Load: Add a new configuration to enable datatype conversion when converting into Tablets & Refactor LoadAnalyzeException. (#14646)
add debdb527d89 Reduced the error log cause by batch activate template when some devices are already activated (#14661)
add a54573a2afb Use DataNodeDevicePathCache when insert by sql (#14662)
add 5f01571f4c6 Pipe: tree/table model isolation for alter/start/stop/drop/show pipe operations & support 'mode.double-living' (#14386)
add 7dc5179def6 Fix select-distinct when there is no order by and select items contains expression
add e92be6f63ab [Table Model Subquery] Support uncorrelated in predicate (#14438)
add bdf36ba1907 Skip subquery plan if subqueryExpression List is empty for efficiency (#14647)
add 2d436e06897 Optimized the error code when table/column does not exist & Reduced some unnecessary IT execution
add 04df3e279a2 Fixed the un-renamed ID/MEASUREMENTs in some non-runnable ITs
add 7f0ac972839 Pipe: Fix NPE caused by forced type conversion and fix IoTDBipipeTypeConversionISessionIT of table model (#14667)
add a5e13fd8f7c Pipe: report linked tsfile size & Subscription: decrease reference count for other enriched events & add logging to observe possible stuck situations (#14668)
add c8d7c5dfea9 Reopen drop column and delete devices operations
add 937df761e31 IoTConsensusV2: Fix some consensus group missing due to recover failed or blocked (#14613)
add 5bd1eadea76 Clone value array when insert non aligned tablet with null (#14643)
add 01e1420fa0a Add Memory Threshold (#14597)
add 82c85483572 Ignore the DatabaseNotExistsException for adjustMaxRegionGroupNum when querying dataRegionGroupCount (#14669)
add 3ac203c4e79 Pipe: Implement table meta/deletion sync & Fix the bug that delete data node's pattern is not parsed / pipe transferred delete data node's progressIndex is not binded & Implement mem control for schema snapshot parser (#14156)
add b349f565fd8 Pipe/Load: Fixed the issue where the TsFileResource of the PlainDeviceTimeIndex type was judged to be empty (triggered by IoTDB loading TsFiles in v3 and the tsfiles are syned to other clusters) (#14692)
add 97a08c880fb Pipe Plugin: intro class annotations for pipe plugin & tree/table model isolation for drop/show pipe plugin operations (#14673)
add f2e20b54408 Modify tablet usage & don't cache TableSchema map in default option (#14635)
add f129298c9e8 Pipe: Fix PipeWriteBackSink using toTPipeTransferReq causing NPE due to uninitialized buffer (#14672)
add 701a5156e7a Pipe: Optimize the process of collecting PipeTsFileInsertionEvent in table model to reduce unnecessary parsing of TsFile (#14697)
add b4edfdf1d68 IoTConsensusV2: Accelerate follower receive dispatch based on TPCx case
add b51f63c7108 Load: Fix some issues during loading tsfiles of old version (#14649)
add c65b8264156 Pipe: strict check for synonym pipe parameters to avoid ambiguity (#14694)
add a11401b494f add repair data for table model (#14705)
add c3df7d54496 Renaming the IT Methods of the IoTDB REST Service Table Model (#14681)
add 6cb9035e8f6 Pipe IT: Added flush for assertData on targetEnv to avoid IoTV2 batch mode lead to large delay (#14707)
add 42878058f11 [RATISCONSENSUS] Bump ratis version to 3.1.3 (#14709)
add 6961e2154a7 Fix NPE when concurrent closing mod files (#14708)
add 8e7605f80db Fix cache invalidation failure and add retry (#14711)
add 44e9725ab94 Publish build scans to develocity.apache.org (#14701)
add c3baa7a4978 Subscription: fix `hasNext` method in tablet batch to support multiple tsfile in the same batch (#14715)
add 6036cabd9dc Implemented access control for pipe/subscription/table/database
add 3072ae5a668 Implement distinct in AggFunction
add 49a91ecd348 Add more user-friendly error messages for the last and last_by aggregation
add 141b7ba634c Extend Region & Remove Region & Reconstruct Region (#14634)
add a011b013711 Pipe: Unify the case of the table model database and fix the incorrect use of iterators in Load File (#14718)
add 75dda9fb0ef Pipe: Fixed the bug that alter logical view statement uses a common partial path (#14713)
add bc5fdaed8d9 Bump TsFile version to 2.0.0-250118-SNAPSHOT
add 670b45675e0 Pipe: Fix HA issues caused by exceptions not handled in handshake (IoTDBAirGapConnector / IoTDBSyncClientManager) (#14706)
add fea2e4fc8f8 [Table Model Subquery] Support uncorrelated quantified comparison
add 263443dcb53 The time range of device in resource file may be larger than the actual time range in tsfile (#14712)
new 602a726ae63 merge master
new b9dd299cbe8 adjust api
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/cluster-it-1c1d.yml | 2 +-
.github/workflows/cluster-it-1c1d1a.yml | 2 +-
.github/workflows/cluster-it-1c3d.yml | 2 +-
.github/workflows/compile-check.yml | 2 +-
.github/workflows/daily-it.yml | 2 +-
.github/workflows/daily-ut.yml | 2 +-
.github/workflows/dependency-check.yml | 2 +-
.github/workflows/pipe-it-2cluster.yml | 2 +-
.github/workflows/sonar-codecov.yml | 2 +-
.github/workflows/table-cluster-it-1c1d.yml | 2 +-
.github/workflows/table-cluster-it-1c3d.yml | 2 +-
.github/workflows/unit-test.yml | 2 +-
.github/workflows/vulnerability-check.yml | 2 +-
.mvn/develocity.xml | 3 +-
.mvn/extensions.xml | 4 +-
.../java/org/apache/iotdb/CountPointProcessor.java | 2 +
.../iotdb/AlignedTimeseriesSessionExample.java | 67 +-
.../org/apache/iotdb/SessionConcurrentExample.java | 23 -
.../main/java/org/apache/iotdb/SessionExample.java | 107 +---
.../org/apache/iotdb/trigger/LoggerTrigger.java | 12 +-
.../iotdb/trigger/StatisticsUpdaterTrigger.java | 4 +-
.../apache/iotdb/udf/AggregateFunctionExample.java | 35 +-
.../apache/iotdb/udf/ScalarFunctionExample.java | 57 +-
.../db/query/udf/example/relational/AllSum.java | 49 +-
.../query/udf/example/relational/ContainNull.java | 21 +-
.../relational/{DatePlusOne.java => DatePlus.java} | 31 +-
.../query/udf/example/relational/FirstTwoSum.java | 33 +-
.../db/query/udf/example/relational/MyAvg.java | 32 +-
.../db/query/udf/example/relational/MyCount.java | 19 +-
.../java/org/apache/iotdb/it/env/EnvFactory.java | 5 +
.../iotdb/it/env/cluster/env/AbstractEnv.java | 49 ++
.../iotdb/it/env/cluster/node/AINodeWrapper.java | 5 +
.../it/env/cluster/node/AbstractNodeWrapper.java | 12 +
.../it/env/cluster/node/ConfigNodeWrapper.java | 5 +
.../iotdb/it/env/cluster/node/DataNodeWrapper.java | 5 +
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 12 +
.../org/apache/iotdb/it/utils/TsFileGenerator.java | 87 +--
.../iotdb/it/utils/TsFileTableGenerator.java | 186 ++++++
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 7 +
...ionMigrateDataNodeCrashITFrameworkForIoTV1.java | 2 +-
...ionMigrateDataNodeCrashITFrameworkForIoTV2.java | 2 +-
...oTDBRegionOperationReliabilityITFramework.java} | 174 ++++--
.../IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java | 157 +++++
.../commit/IoTDBRegionReconstructForIoTV1IT.java | 141 +++++
.../IoTDBRegionMigrateNormalITForIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateOtherITForIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateNormalITForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateOtherITForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateDataNodeCrashForIoTV1IT.java | 4 +-
...TDBRegionMigrateDataNodeCrashForIoTV2Batch.java | 4 +-
...DBRegionMigrateDataNodeCrashForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV1IT.java | 4 +-
.../IoTDBRegionMigrateConfigNodeCrashIoTV1IT.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV2Batch.java | 4 +-
...oTDBRegionMigrateConfigNodeCrashIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV2Stream.java | 4 +-
...TDBRegionMigrateConfigNodeCrashIoTV2Stream.java | 4 +-
.../IoTDBRemoveDataNodeITFramework.java | 22 +-
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 80 ++-
.../iotdb/db/it/IoTDBLoadTsFileWithModIT.java | 172 ++++++
.../it/IoTDBSyntaxConventionStringLiteralIT.java | 3 +-
.../org/apache/iotdb/db/it/utils/TestUtils.java | 279 ++++++++-
.../iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java | 3 +-
.../pipe/it/autocreate/IoTDBPipeAutoDropIT.java | 10 +-
.../pipe/it/autocreate/IoTDBPipeClusterIT.java | 98 +++
.../pipe/it/autocreate/IoTDBPipeIdempotentIT.java | 93 +++
.../pipe/it/autocreate/IoTDBPipeNullValueIT.java | 28 +-
.../it/autocreate/IoTDBPipeSwitchStatusIT.java | 23 +-
.../pipe/it/manual/IoTDBPipeReqAutoSliceIT.java | 8 +-
...nclusionIT.java => IoTDBPipeTableManualIT.java} | 222 ++++---
.../manual/IoTDBPipeTypeConversionISessionIT.java | 10 +-
.../iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java | 30 +-
.../pipe/it/tablemodel/IoTDBPipeAutoDropIT.java | 10 +-
.../pipe/it/tablemodel/IoTDBPipeClusterIT.java | 65 ++
.../pipe/it/tablemodel/IoTDBPipeDataSinkIT.java | 450 +++++++++++++-
.../it/tablemodel/IoTDBPipeDoubleLivingIT.java | 329 ++++++++++
.../pipe/it/tablemodel/IoTDBPipeIsolationIT.java | 377 ++++++++++++
.../pipe/it/tablemodel/IoTDBPipeNullValueIT.java | 4 +-
.../pipe/it/tablemodel/IoTDBPipeProtocolIT.java | 1 -
.../it/tablemodel/IoTDBPipeSwitchStatusIT.java | 23 +-
.../pipe/it/tablemodel/IoTDBPipeSyntaxIT.java | 35 +-
.../IoTDBPipeTypeConversionISessionIT.java | 97 ++-
.../pipe/it/tablemodel/IoTDBPipeWithLoadIT.java | 8 +-
.../iotdb/pipe/it/tablemodel/TableModelUtils.java | 614 +++++++++++++++----
.../relational/it/db/it/IoTDBDeletionTableIT.java | 2 +-
.../relational/it/db/it/IoTDBInsertTableIT.java | 2 +-
.../it/db/it/udf/IoTDBSQLFunctionManagementIT.java | 19 +-
.../udf/IoTDBUserDefinedAggregateFunctionIT.java | 53 +-
.../it/udf/IoTDBUserDefinedScalarFunctionIT.java | 2 +-
.../IoTDBAlignByDeviceWithTemplateTableIT.java | 4 +-
.../scalar/IoTDBFormatFunctionTableIT.java | 21 +-
...oTDBNoSelectExpressionAfterAnalyzedTableIT.java | 6 +-
.../query/old/query/IoTDBNullValueFillTableIT.java | 327 ----------
.../it/query/old/query/IoTDBResultSetTableIT.java | 2 +-
.../query/old/query/IoTDBSelectSchemaTableIT.java | 22 +-
.../it/query/recent/IoTDBTableAggregationIT.java | 656 +++++++++++++++++++-
.../recent/subquery/SubqueryDataSetUtils.java | 12 +-
.../IoTDBUncorrelatedInPredicateSubqueryIT.java | 319 ++++++++++
.../IoTDBUncorrelatedQuantifiedComparisonIT.java | 674 +++++++++++++++++++++
.../IoTDBUncorrelatedScalarSubqueryIT.java | 4 +-
.../it/rest/it/IoTDBRestServiceCaseWhenThenIT.java | 7 +-
.../it/rest/it/IoTDBRestServiceFlushQueryIT.java | 6 +-
.../relational/it/rest/it/IoTDBRestServiceIT.java | 3 +-
...IT.java => IoTDBRestServiceInsertValuesIT.java} | 20 +-
.../relational/it/schema/IoTDBDatabaseIT.java | 41 ++
.../iotdb/relational/it/schema/IoTDBDeviceIT.java | 47 +-
.../iotdb/relational/it/schema/IoTDBTableIT.java | 122 ++--
.../it/session/IoTDBSessionRelationalIT.java | 29 +-
.../iotdb/session/it/IoTDBSessionComplexIT.java | 8 +-
.../iotdb/session/it/IoTDBSessionSimpleIT.java | 78 ++-
.../session/it/IoTDBSessionSyntaxConventionIT.java | 18 +-
.../org/apache/iotdb/session/it/SessionIT.java | 32 +-
.../it/local/IoTDBSubscriptionDataTypeIT.java | 8 +-
.../java/org/apache/iotdb/util/MagicUtils.java | 61 ++
.../java/org/apache/iotdb/pipe/api/PipePlugin.java | 37 ++
.../iotdb/pipe/api/annotation/TableModel.java | 18 +-
.../iotdb/pipe/api/annotation/TreeModel.java | 18 +-
.../parameter/PipeParameterValidator.java | 42 +-
.../analysis/AggregateFunctionAnalysis.java | 62 ++
.../api/customizer/analysis/FunctionAnalysis.java | 8 +-
.../ScalarFunctionAnalysis.java} | 26 +-
...ctionParameters.java => FunctionArguments.java} | 28 +-
...tion.java => UDFArgumentNotValidException.java} | 4 +-
.../udf/api/relational/AggregateFunction.java | 48 +-
.../iotdb/udf/api/relational/ScalarFunction.java | 47 +-
.../iotdb/udf/api/relational/TableFunction.java | 51 +-
.../relational/table/TableFunctionAnalysis.java | 4 +-
.../relational/table/argument/DescribedSchema.java | 20 +-
.../api/relational/table/argument/Descriptor.java | 95 ---
.../table/argument/DescriptorArgument.java | 57 --
.../relational/table/argument/TableArgument.java | 11 -
.../processor/TableFunctionDataProcessor.java | 32 +-
.../processor/TableFunctionLeafProcessor.java | 5 +
.../DescriptorParameterSpecification.java | 57 --
.../specification/ReturnTypeSpecification.java | 68 ---
.../specification/TableParameterSpecification.java | 9 -
.../apache/iotdb/udf/api/type/ColumnCategory.java | 37 --
.../apache/iotdb/tool/data/AbstractDataTool.java | 32 +-
.../org/apache/iotdb/tool/data/ExportData.java | 43 +-
.../org/apache/iotdb/tool/data/ImportData.java | 24 +-
.../org/apache/iotdb/tool/tsfile/ExportTsFile.java | 3 -
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 3 +
.../java/org/apache/iotdb/session/Session.java | 27 +-
.../payload/SubscriptionFileHandler.java | 24 +-
.../payload/SubscriptionSessionDataSet.java | 12 +-
.../org/apache/iotdb/session/util/RetryUtils.java | 37 +-
.../apache/iotdb/session/util/SessionUtils.java | 25 +-
.../java/org/apache/iotdb/session/TabletTest.java | 28 +-
.../iotdb/session/util/SessionUtilsTest.java | 16 +-
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 16 +-
.../antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4 | 8 +
.../consensus/request/ConfigPhysicalPlan.java | 10 +-
.../consensus/request/ConfigPhysicalPlanType.java | 4 +
.../request/ConfigPhysicalPlanVisitor.java | 110 +++-
.../DescTable4InformationSchemaPlan.java} | 9 +-
.../ShowTable4InformationSchemaPlan.java} | 8 +-
.../payload/PipeCreateTablePlan.java} | 15 +-
.../pipe/payload/PipeDeactivateTemplatePlan.java | 22 +-
.../write/pipe/payload/PipeDeleteDevicesPlan.java | 130 ++++
.../request/write/table/PreCreateTablePlan.java | 11 +-
.../response/pipe/plugin/PipePluginTableResp.java | 27 +-
.../response/pipe/task/PipeTableResp.java | 7 +
...p.java => DescTable4InformationSchemaResp.java} | 20 +-
...p.java => ShowTable4InformationSchemaResp.java} | 17 +-
.../iotdb/confignode/manager/ClusterManager.java | 2 +-
.../iotdb/confignode/manager/ConfigManager.java | 66 +-
.../apache/iotdb/confignode/manager/IManager.java | 42 +-
.../iotdb/confignode/manager/ProcedureManager.java | 491 +++++++++++----
.../payload/PipeTransferConfigSnapshotSealReq.java | 32 +-
.../protocol/IoTDBConfigRegionAirGapConnector.java | 8 +
.../protocol/IoTDBConfigRegionConnector.java | 10 +-
.../coordinator/plugin/PipePluginCoordinator.java | 29 +
.../pipe/coordinator/task/PipeTaskCoordinator.java | 74 ++-
.../pipe/event/PipeConfigRegionSnapshotEvent.java | 3 +-
.../extractor/ConfigRegionListeningFilter.java | 89 +--
.../pipe/extractor/ConfigRegionListeningQueue.java | 23 +-
.../pipe/extractor/IoTDBConfigRegionExtractor.java | 74 ++-
...ConfigPhysicalPlanTablePatternParseVisitor.java | 135 +++++
...ConfigPhysicalPlanTreePatternParseVisitor.java} | 6 +-
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 170 +++++-
.../PipeConfigPhysicalPlanTSStatusVisitor.java | 76 +++
.../manager/schema/ClusterSchemaManager.java | 158 +++--
.../persistence/executor/ConfigPlanExecutor.java | 5 +
.../confignode/persistence/pipe/PipeInfo.java | 2 +-
.../persistence/pipe/PipePluginInfo.java | 39 +-
.../confignode/persistence/pipe/PipeTaskInfo.java | 28 +-
.../schema/CNPhysicalPlanGenerator.java | 77 ++-
.../persistence/schema/ClusterSchemaInfo.java | 77 +++
.../confignode/persistence/schema/ConfigMTree.java | 29 +-
.../schema/ConfignodeSnapshotParser.java | 30 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 23 +-
.../procedure/env/RegionMaintainHandler.java | 13 +-
.../impl/pipe/task/AlterPipeProcedureV2.java | 4 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 42 +-
.../impl/region/AddRegionPeerProcedure.java | 84 +--
.../impl/region/ReconstructRegionProcedure.java | 209 +++++++
.../impl/region/RegionMigrateProcedure.java | 39 +-
.../impl/region/RegionOperationProcedure.java} | 33 +-
.../impl/region/RemoveRegionPeerProcedure.java | 52 +-
.../impl/schema/DeleteDatabaseProcedure.java | 2 +-
.../impl/schema/DeleteTimeSeriesProcedure.java | 4 +-
.../table/AbstractAlterOrDropTableProcedure.java | 10 +-
.../impl/schema/table/AddTableColumnProcedure.java | 16 +-
.../impl/schema/table/CreateTableProcedure.java | 20 +-
.../impl/schema/table/DeleteDevicesProcedure.java | 43 +-
.../schema/table/DropTableColumnProcedure.java | 21 +-
.../impl/schema/table/DropTableProcedure.java | 24 +-
.../schema/table/RenameTableColumnProcedure.java | 18 +-
.../schema/table/SetTablePropertiesProcedure.java | 18 +-
.../impl/trigger/CreateTriggerProcedure.java | 9 +-
...itionState.java => ReconstructRegionState.java} | 8 +-
.../procedure/store/ProcedureFactory.java | 40 +-
.../confignode/procedure/store/ProcedureType.java | 12 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 52 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 40 ++
.../response/pipe/PipePluginTableRespTest.java | 4 +-
.../connector/PipeConfigNodeThriftRequestTest.java | 12 +-
...igPhysicalPlanTablePatternParseVisitorTest.java | 133 ++++
...igPhysicalPlanTreePatternParseVisitorTest.java} | 44 +-
.../pipe/annotation/PipePluginAnnotationTest.java} | 20 +-
.../pipe/receiver/PipeEnrichedProcedureTest.java | 194 ++++++
.../schema/table/AddTableColumnProcedureTest.java | 8 +-
.../schema/table/CreateTableProcedureTest.java | 5 +-
.../schema/table/DeleteDevicesProcedureTest.java | 5 +-
.../schema/table/DropTableColumnProcedureTest.java | 4 +-
.../impl/schema/table/DropTableProcedureTest.java | 4 +-
.../table/RenameTableColumnProcedureTest.java | 4 +-
.../table/SetTablePropertiesProcedureTest.java | 6 +-
.../apache/iotdb/consensus/pipe/PipeConsensus.java | 57 +-
.../apache/iotdb/consensus/ratis/RatisClient.java | 6 +-
.../apache/iotdb/consensus/ratis/utils/Utils.java | 1 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 4 +-
.../dataregion/DataExecutionVisitor.java | 9 +-
.../schemaregion/SchemaExecutionVisitor.java | 15 +-
.../schemaregion/SchemaRegionStateMachine.java | 15 +-
.../LoadAnalyzeException.java} | 8 +-
.../LoadAnalyzeTableColumnDisorderException.java} | 7 +-
.../LoadAnalyzeTypeMismatchException.java} | 6 +-
.../pipe/agent/plugin/PipeDataNodePluginAgent.java | 20 +-
.../dataregion/PipeDataRegionPluginAgent.java | 49 ++
.../agent/task/connection/PipeEventCollector.java | 35 +-
.../client/IoTDBDataNodeAsyncClientManager.java | 2 +-
.../batch/PipeTabletEventTsFileBatch.java | 384 +++---------
.../PipeTransferDataNodeHandshakeV1Req.java | 10 +-
.../evolvable/request/PipeTransferPlanNodeReq.java | 18 +-
.../request/PipeTransferSchemaSnapshotSealReq.java | 94 ++-
.../request/PipeTransferTabletInsertNodeReqV2.java | 2 +-
.../request/PipeTransferTabletRawReq.java | 4 +-
.../request/PipeTransferTabletRawReqV2.java | 9 +-
.../request/PipeTransferTsFileSealWithModReq.java | 42 +-
.../airgap/IoTDBDataNodeAirGapConnector.java | 4 +
.../airgap/IoTDBDataRegionAirGapConnector.java | 4 +
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 19 +-
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 2 +
.../connector/protocol/opcua/OpcUaConnector.java | 4 +
.../connector/protocol/opcua/OpcUaNameSpace.java | 44 +-
.../pipeconsensus/PipeConsensusAsyncConnector.java | 4 +
.../pipeconsensus/PipeConsensusSyncConnector.java | 4 +
.../request/PipeConsensusDeleteNodeReq.java | 4 +-
.../async/IoTDBDataRegionAsyncConnector.java | 14 +-
.../PipeTransferTabletBatchEventHandler.java | 2 +-
.../async/handler/PipeTransferTsFileHandler.java | 13 +-
.../thrift/sync/IoTDBDataNodeSyncConnector.java | 4 +
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 23 +-
.../thrift/sync/IoTDBSchemaRegionConnector.java | 26 +-
.../protocol/websocket/WebSocketConnector.java | 2 +
.../protocol/writeback/WriteBackConnector.java | 199 +++++-
.../util/builder/PipeTableModeTsFileBuilder.java | 273 +++++++++
.../util/builder/PipeTreeModelTsFileBuilder.java | 268 ++++++++
.../connector/util/builder/PipeTsFileBuilder.java | 162 +++++
.../util/{ => cacher}/LeaderCacheUtils.java | 2 +-
.../sorter/PipeTableModelTabletEventSorter.java | 273 +++++++++
.../util/{ => sorter}/PipeTabletEventSorter.java | 107 +---
.../sorter/PipeTreeModelTabletEventSorter.java | 123 ++++
.../pipe/consensus/deletion/DeletionResource.java | 2 +-
.../db/pipe/event/common/PipeInsertionEvent.java | 15 +-
.../common/deletion/PipeDeleteDataNodeEvent.java | 24 +-
.../db/pipe/event/common/row/PipeRowCollector.java | 2 +-
.../schema/PipeSchemaRegionSnapshotEvent.java | 75 ++-
.../schema/PipeSchemaSerializableEventType.java | 15 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 2 +-
.../tablet/parser/TabletInsertionEventParser.java | 8 +-
.../TabletInsertionEventTreePatternParser.java | 13 +-
.../scan/TsFileInsertionEventScanParser.java | 42 +-
.../table/TsFileInsertionEventTableParser.java | 2 +-
.../dataregion/DataRegionListeningFilter.java | 21 +-
.../dataregion/IoTDBDataRegionExtractor.java | 156 ++---
...oricalDataRegionTsFileAndDeletionExtractor.java | 4 +-
.../listener/PipeInsertionDataNodeListener.java | 27 +-
.../schemaregion/IoTDBSchemaRegionExtractor.java | 28 +-
.../PipePlanTablePatternParseVisitor.java | 77 +++
...r.java => PipePlanTreePatternParseVisitor.java} | 7 +-
.../schemaregion/SchemaRegionListeningFilter.java | 87 +--
.../schemaregion/SchemaRegionListeningQueue.java | 8 +-
.../pipe/metric/PipeDataNodeReceiverMetrics.java | 48 +-
.../processor/aggregate/AggregateProcessor.java | 2 +
.../StandardStatisticsOperatorProcessor.java | 2 +
.../processor/TumblingWindowingProcessor.java | 2 +
.../changing/ChangingValueSamplingProcessor.java | 2 +
.../sdt/SwingingDoorTrendingSamplingProcessor.java | 2 +
.../tumbling/TumblingTimeSamplingProcessor.java | 2 +
.../pipeconsensus/PipeConsensusProcessor.java | 4 +
.../schemachange/RenameDatabaseProcessor.java | 11 +-
.../twostage/plugin/TwoStageCountProcessor.java | 37 +-
.../pipeconsensus/PipeConsensusReceiver.java | 33 +-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 223 +++++--
.../statement/PipeConvertedInsertRowStatement.java | 30 +-
.../PipeConvertedInsertTabletStatement.java | 52 +-
.../visitor/PipePlanToStatementVisitor.java | 54 +-
.../PipeStatementTablePatternParseVisitor.java} | 24 +-
...a => PipeStatementTreePatternParseVisitor.java} | 6 +-
...leStatementDataTypeConvertExecutionVisitor.java | 10 +-
...eeStatementDataTypeConvertExecutionVisitor.java | 3 +-
...r.java => PipeTreeStatementToBatchVisitor.java} | 2 +-
.../pipe/resource/memory/PipeMemoryWeightUtil.java | 21 +-
.../resource/tsfile/PipeTsFileResourceManager.java | 5 +-
.../iotdb/db/protocol/client/ConfigNodeClient.java | 57 ++
.../impl/DataNodeInternalRPCServiceImpl.java | 2 +
.../execution/executor/RegionWriteExecutor.java | 55 +-
.../function/table/ExcludeColumnFunction.java | 32 +-
.../execution/function/table/HOPTableFunction.java | 55 +-
.../execution/function/table/SplitFunction.java | 6 +-
.../operator/process/EnforceSingleRowOperator.java | 5 +-
.../process/function/PartitionRecognizer.java | 124 +---
.../process/function/TableFunctionOperator.java | 73 ++-
.../process/function/partition/PartitionState.java | 22 +-
.../operator/process/function/partition/Slice.java | 153 +++++
.../partition/SliceCache.java} | 41 +-
.../merge/comparator/JoinKeyComparatorFactory.java | 2 +-
.../schema/source/TableDeviceQuerySource.java | 4 +-
.../InformationSchemaContentSupplierFactory.java | 372 ++++++++++--
...perator.java => MergeSortSemiJoinOperator.java} | 130 ++--
.../relational/aggregation/AccumulatorFactory.java | 265 +++++++-
.../relational/aggregation/AggregationMask.java | 204 +++++++
.../relational/aggregation/AvgAccumulator.java | 122 +++-
.../relational/aggregation/CountAccumulator.java | 22 +-
...IfAccumulator.java => CountAllAccumulator.java} | 48 +-
.../relational/aggregation/CountIfAccumulator.java | 22 +-
.../relational/aggregation/ExtremeAccumulator.java | 94 ++-
.../relational/aggregation/FirstAccumulator.java | 158 +++--
.../relational/aggregation/FirstByAccumulator.java | 164 +++--
.../aggregation/FirstByDescAccumulator.java | 132 +++-
.../aggregation/FirstDescAccumulator.java | 126 +++-
.../relational/aggregation/LastAccumulator.java | 140 ++++-
.../relational/aggregation/LastByAccumulator.java | 146 ++++-
.../aggregation/LastByDescAccumulator.java | 150 ++++-
.../aggregation/LastDescAccumulator.java | 144 ++++-
.../MaskedRecordIterator.java} | 24 +-
.../relational/aggregation/MaxAccumulator.java | 140 ++++-
.../relational/aggregation/MinAccumulator.java | 140 ++++-
.../relational/aggregation/SumAccumulator.java | 114 +++-
.../relational/aggregation/TableAccumulator.java | 2 +-
.../relational/aggregation/TableAggregator.java | 5 +-
.../aggregation/TableMaxMinByBaseAccumulator.java | 146 +++--
.../aggregation/TableModeAccumulator.java | 192 ++++--
.../aggregation/TableVarianceAccumulator.java | 158 +++--
.../UserDefinedAggregateFunctionAccumulator.java | 35 +-
.../aggregation/grouped/GroupedAccumulator.java | 4 +-
.../aggregation/grouped/GroupedAggregator.java | 6 +-
.../aggregation/grouped/GroupedAvgAccumulator.java | 119 +++-
.../grouped/GroupedCountAccumulator.java | 22 +-
.../grouped/GroupedCountIfAccumulator.java | 22 +-
.../grouped/GroupedExtremeAccumulator.java | 95 ++-
.../grouped/GroupedFirstAccumulator.java | 153 ++++-
.../grouped/GroupedFirstByAccumulator.java | 191 ++++--
.../grouped/GroupedLastAccumulator.java | 133 +++-
.../grouped/GroupedLastByAccumulator.java | 150 ++++-
.../aggregation/grouped/GroupedMaxAccumulator.java | 141 ++++-
.../grouped/GroupedMaxMinByBaseAccumulator.java | 147 ++++-
.../aggregation/grouped/GroupedMinAccumulator.java | 141 ++++-
.../grouped/GroupedModeAccumulator.java | 246 ++++++--
.../aggregation/grouped/GroupedSumAccumulator.java | 111 +++-
.../GroupedUserDefinedAggregateAccumulator.java | 14 +-
.../grouped/GroupedVarianceAccumulator.java | 159 +++--
.../aggregation/grouped/hash/GroupByHash.java | 2 +
.../aggregation/grouped/hash/MarkDistinctHash.java | 90 +++
.../relational/ColumnTransformerBuilder.java | 20 +-
.../iotdb/db/queryengine/plan/Coordinator.java | 8 +
.../db/queryengine/plan/analyze/AnalyzeUtils.java | 9 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 30 +-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 27 +-
.../analyze/load/LoadTsFileTableSchemaCache.java | 17 +-
.../load/LoadTsFileToTableModelAnalyzer.java | 34 +-
.../load/LoadTsFileToTreeModelAnalyzer.java | 22 +-
.../analyze/load/LoadTsFileTreeSchemaCache.java | 3 +-
.../load/TreeSchemaAutoCreatorAndVerifier.java | 30 +-
.../execution/config/TableConfigTaskVisitor.java | 141 ++++-
.../execution/config/TreeConfigTaskVisitor.java | 30 +-
.../config/executor/ClusterConfigTaskExecutor.java | 159 ++++-
.../config/executor/IConfigTaskExecutor.java | 19 +-
.../config/metadata/DropPipePluginTask.java | 13 +-
.../config/metadata/ShowPipePluginsTask.java | 20 +-
.../ExtendRegionTask.java} | 14 +-
.../metadata/{ => region}/MigrateRegionTask.java | 4 +-
.../ReconstructRegionTask.java} | 16 +-
.../RemoveRegionTask.java} | 14 +-
.../config/metadata/relational/ShowDBTask.java | 4 +-
.../metadata/relational/ShowTablesDetailsTask.java | 47 +-
.../config/metadata/relational/ShowTablesTask.java | 35 +-
.../execution/config/sys/pipe/AlterPipeTask.java | 8 +-
.../execution/config/sys/pipe/DropPipeTask.java | 7 +-
.../execution/config/sys/pipe/ShowPipeTask.java | 3 +-
.../execution/config/sys/pipe/StartPipeTask.java | 7 +-
.../execution/config/sys/pipe/StopPipeTask.java | 7 +-
.../db/queryengine/plan/parser/ASTVisitor.java | 38 +-
.../plan/planner/LogicalPlanVisitor.java | 7 +-
.../plan/planner/TableOperatorGenerator.java | 102 +++-
.../plan/planner/plan/node/PlanGraphPrinter.java | 34 +-
.../plan/planner/plan/node/PlanNodeType.java | 4 +
.../plan/planner/plan/node/PlanVisitor.java | 5 +
.../node/metadata/write/AlterTimeSeriesNode.java | 9 +-
.../node/metadata/write/CreateTimeSeriesNode.java | 74 +--
.../plan/node/pipe/PipeEnrichedDeleteDataNode.java | 66 +-
.../plan/node/pipe/PipeEnrichedInsertNode.java | 44 +-
.../plan/node/pipe/PipeEnrichedWritePlanNode.java | 23 +-
.../relational/analyzer/ExpressionAnalyzer.java | 37 +-
.../plan/relational/analyzer/Scope.java | 8 +-
.../relational/analyzer/StatementAnalyzer.java | 235 +++----
.../plan/relational/metadata/Metadata.java | 4 +-
.../relational/metadata/TableMetadataImpl.java | 53 +-
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 14 +-
.../fetcher/TableHeaderSchemaValidator.java | 11 +-
.../plan/relational/planner/IrTypeAnalyzer.java | 15 +-
.../plan/relational/planner/QueryPlanner.java | 2 +-
.../plan/relational/planner/RelationPlanner.java | 91 +--
.../relational/planner/SimplePlanRewriter.java | 88 +++
.../plan/relational/planner/SubqueryPlanner.java | 84 ++-
.../relational/planner/TableLogicalPlanner.java | 15 +-
.../distribute/TableDistributedPlanGenerator.java | 24 +
.../planner/iterative/rule/PruneApplyColumns.java | 138 +++++
...Correlation.java => PruneApplyCorrelation.java} | 38 +-
.../iterative/rule/PruneApplySourceColumns.java | 95 +++
...ava => RemoveUnreferencedScalarApplyNodes.java} | 19 +-
.../rule/RemoveUnreferencedScalarSubqueries.java | 70 +++
.../TransformFilteringSemiJoinToInnerJoin.java | 150 +++++
...mUncorrelatedInPredicateSubqueryToSemiJoin.java | 95 +++
.../plan/relational/planner/node/Patterns.java | 7 +-
.../plan/relational/planner/node/SemiJoinNode.java | 181 ++++++
.../node/schema/CreateOrUpdateTableDeviceNode.java | 14 +-
.../DataNodeLocationSupplierFactory.java | 11 +
.../optimizations/LogicalOptimizeFactory.java | 26 +-
.../PushAggregationIntoTableScan.java | 4 +
.../optimizations/PushPredicateIntoTableScan.java | 235 ++++++-
...mQuantifiedComparisonApplyToCorrelatedJoin.java | 341 +++++++++++
.../optimizations/UnaliasSymbolReferences.java | 29 +
.../sql/ast/AbstractQueryDeviceWithCache.java | 2 +
.../relational/sql/ast/AbstractTraverseDevice.java | 8 +
.../plan/relational/sql/ast/AlterPipe.java | 22 +-
.../plan/relational/sql/ast/AstVisitor.java | 27 +-
.../{DescribeTable.java => CountStatement.java} | 40 +-
.../relational/sql/ast/CreateOrUpdateDevice.java | 31 +-
.../plan/relational/sql/ast/CreatePipe.java | 4 +-
.../plan/relational/sql/ast/Delete.java | 28 +-
.../plan/relational/sql/ast/DeleteDevice.java | 2 -
.../plan/relational/sql/ast/Descriptor.java | 80 ---
.../plan/relational/sql/ast/DescriptorField.java | 88 ---
.../plan/relational/sql/ast/DropColumn.java | 3 -
.../plan/relational/sql/ast/DropPipe.java | 10 +-
.../plan/relational/sql/ast/LoadTsFile.java | 9 +
.../plan/relational/sql/ast/PipeEnriched.java | 8 +-
.../plan/relational/sql/ast/ShowPipes.java | 10 +-
.../plan/relational/sql/ast/StartPipe.java | 8 +-
...{SetConfiguration.java => StartRepairData.java} | 6 +-
.../plan/relational/sql/ast/Statement.java | 4 +-
.../plan/relational/sql/ast/StopPipe.java | 8 +-
.../{SetConfiguration.java => StopRepairData.java} | 6 +-
.../relational/sql/ast/TableFunctionArgument.java | 5 +-
.../sql/ast/TableFunctionDescriptorArgument.java | 87 ---
.../plan/relational/sql/ast/Update.java | 6 +-
.../plan/relational/sql/parser/AstBuilder.java | 104 ++--
.../plan/relational/sql/rewrite/ShowRewrite.java | 51 +-
.../plan/relational/sql/util/SqlFormatter.java | 23 -
.../plan/scheduler/load/LoadTsFileScheduler.java | 9 +-
.../plan/statement/StatementVisitor.java | 18 +-
.../plan/statement/crud/InsertBaseStatement.java | 22 +-
.../plan/statement/crud/InsertRowStatement.java | 18 +-
.../plan/statement/crud/LoadTsFileStatement.java | 9 +
.../metadata/AlterTimeSeriesStatement.java | 7 +-
.../metadata/pipe/AlterPipeStatement.java | 31 +-
.../metadata/pipe/DropPipePluginStatement.java | 17 +-
.../statement/metadata/pipe/DropPipeStatement.java | 23 +-
.../metadata/pipe/ShowPipePluginsStatement.java | 23 +-
.../metadata/pipe/ShowPipesStatement.java | 18 +-
.../metadata/pipe/StartPipeStatement.java | 17 +-
.../statement/metadata/pipe/StopPipeStatement.java | 17 +-
.../region/ExtendRegionStatement.java} | 32 +-
.../{ => region}/MigrateRegionStatement.java | 2 +-
.../region/ReconstructRegionStatement.java} | 31 +-
.../region/RemoveRegionStatement.java} | 32 +-
.../plan/statement/pipe/PipeEnrichedStatement.java | 10 +-
.../attribute/DeviceAttributeStore.java | 10 +-
.../attribute/IDeviceAttributeStore.java | 6 +-
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 136 +++--
.../schemaregion/mtree/impl/mem/MemMTreeStore.java | 52 +-
.../mtree/impl/mem/mnode/basic/BasicMNode.java | 22 +-
.../impl/mem/snapshot/MemMTreeSnapshotUtil.java | 37 +-
.../schemaengine/table/InformationSchemaUtils.java | 11 +-
.../db/service/metrics/DataNodeMetricsHelper.java | 4 +-
.../metrics/memory/ConsensusMemoryMetrics.java} | 64 +-
.../metrics/memory/GlobalMemoryMetrics.java | 132 ++++
.../metrics/memory/QueryEngineMemoryMetrics.java | 179 ++++++
.../metrics/memory/SchemaEngineMemoryMetrics.java | 124 ++++
.../metrics/memory/StorageEngineMemoryMetrics.java | 210 +++++++
.../metrics/memory/StreamEngineMemoryMetrics.java} | 64 +-
.../metrics/memory/ThresholdMemoryMetrics.java | 194 ------
.../db/storageengine/dataregion/DataRegion.java | 123 ++--
.../task/InsertionCrossSpaceCompactionTask.java | 3 -
.../utils/CompactionTableSchemaCollector.java | 2 +-
.../compaction/repair/RepairDataFileScanUtil.java | 30 +-
.../schedule/CompactionScheduleContext.java | 21 +
.../compaction/schedule/CompactionScheduler.java | 60 +-
.../impl/RewriteCrossSpaceCompactionSelector.java | 11 +-
.../dataregion/memtable/TsFileProcessor.java | 4 +
.../dataregion/modification/ModificationFile.java | 89 ++-
.../modification/PartitionLevelModFileManager.java | 5 +-
.../dataregion/snapshot/SnapshotTaker.java | 2 +-
.../dataregion/tsfile/TsFileResource.java | 10 +-
.../tsfile/timeindex/ArrayDeviceTimeIndex.java | 20 +
.../db/storageengine/load/LoadTsFileManager.java | 23 +-
.../load/active/ActiveLoadDirScanner.java | 27 +-
.../load/active/ActiveLoadTsFileLoader.java | 13 +-
.../LoadConvertedInsertTabletStatement.java | 18 +-
...leStatementDataTypeConvertExecutionVisitor.java | 10 +-
...eeStatementDataTypeConvertExecutionVisitor.java | 3 +-
.../converter/LoadTsFileDataTypeConverter.java | 78 ++-
.../load/splitter/TsFileSplitter.java | 5 +-
.../broker/SubscriptionPrefetchingQueue.java | 2 +
.../db/subscription/event/SubscriptionEvent.java | 21 +-
.../batch/SubscriptionPipeTabletEventBatch.java | 24 +-
.../batch/SubscriptionPipeTsFileEventBatch.java | 11 +-
.../db/tools/schema/SRStatementGenerator.java | 208 +++++--
.../tools/schema/SchemaRegionSnapshotParser.java | 32 +-
.../iotdb/db/utils/constant/SqlConstant.java | 1 +
.../iotdb/db/utils/datastructure/BinaryTVList.java | 6 +-
.../db/utils/datastructure/BooleanTVList.java | 6 +-
.../iotdb/db/utils/datastructure/DoubleTVList.java | 6 +-
.../iotdb/db/utils/datastructure/FloatTVList.java | 6 +-
.../iotdb/db/utils/datastructure/IntTVList.java | 6 +-
.../iotdb/db/utils/datastructure/LongTVList.java | 6 +-
.../connector/PipeDataNodeThriftRequestTest.java | 22 +-
.../PipeStatementTablePatternParseVisitorTest.java | 57 ++
... PipeStatementTreePatternParseVisitorTest.java} | 20 +-
.../pipe/connector/PipeTabletEventSorterTest.java | 263 ++++++--
.../db/pipe/consensus/DeletionResourceTest.java | 39 +-
.../pipe/event/PipeTabletInsertionEventTest.java | 25 +-
.../PipePlanTablePatternParseVisitorTest.java | 129 ++++
...va => PipePlanTreePatternParseVisitorTest.java} | 32 +-
.../annotation/PipePluginAnnotationTest.java} | 20 +-
.../resource/PipeTsFileResourceManagerTest.java | 2 +-
.../process/tvf/TableFunctionOperatorTest.java | 57 +-
.../metadata/write/MetadataWriteNodeSerDeTest.java | 2 +-
.../plan/relational/analyzer/DistinctTest.java | 91 +++
.../plan/relational/planner/SubqueryTest.java | 310 ++++++++++
.../planner/assertions/PlanMatchPattern.java | 17 +
.../{SortMatcher.java => SemiJoinMatcher.java} | 41 +-
.../storageengine/buffer/BloomFilterCacheTest.java | 5 +-
.../storageengine/dataregion/DataRegionTest.java | 24 +-
.../compaction/CompactionValidationTest.java | 7 +-
.../InsertionCrossSpaceCompactionSelectorTest.java | 4 +-
.../cross/InsertionCrossSpaceCompactionTest.java | 64 +-
.../inner/InnerSpaceCompactionSelectorTest.java | 2 +-
.../repair/RepairDataFileScanUtilTest.java | 2 +-
.../TableModelFastCompactionPerformerTest.java | 13 +-
...TableModelReadChunkCompactionPerformerTest.java | 13 +-
...TableModelReadPointCompactionPerformerTest.java | 13 +-
.../modification/ModificationFileTest.java | 90 ++-
.../reader/chunk/MemAlignedChunkLoaderTest.java | 2 +-
.../read/reader/chunk/MemChunkLoaderTest.java | 12 +-
.../file/UnsealedTsFileRecoverPerformerTest.java | 4 +-
.../iotdb/db/tools/TsFileAndModSettleToolTest.java | 2 +-
.../iotdb/db/tools/TsFileSelfCheckToolTest.java | 6 +-
.../iotdb/db/tools/TsFileSketchToolTest.java | 10 +-
.../db/utils/SchemaRegionSnapshotParserTest.java | 200 ++++--
iotdb-core/node-commons/pom.xml | 4 +
.../conf/iotdb-system.properties.template | 6 +-
.../exception}/table/ColumnNotExistsException.java | 2 +-
.../table/TableAlreadyExistsException.java | 2 +-
.../exception}/table/TableNotExistsException.java | 2 +-
.../commons/pipe/agent/plugin/PipePluginAgent.java | 17 +-
.../builtin/connector/PlaceholderConnector.java | 4 +
.../connector/donothing/DoNothingConnector.java | 4 +
.../iotdb/airgap/IoTDBAirGapConnector.java | 4 +
.../consensus/PipeConsensusAsyncConnector.java | 4 +
.../iotdb/thrift/IoTDBLegacyPipeConnector.java | 2 +
.../iotdb/thrift/IoTDBThriftAsyncConnector.java | 5 +
.../iotdb/thrift/IoTDBThriftConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSslConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSyncConnector.java | 5 +
.../builtin/connector/opcua/OpcUaConnector.java | 4 +
.../connector/websocket/WebSocketConnector.java | 2 +
.../connector/writeback/WriteBackConnector.java | 4 +
.../extractor/donothing/DoNothingExtractor.java | 4 +
.../builtin/extractor/iotdb/IoTDBExtractor.java | 4 +
.../processor/donothing/DoNothingProcessor.java | 4 +
.../throwing/ThrowingExceptionProcessor.java | 4 +
.../agent/plugin/meta/PipePluginMetaKeeper.java | 47 +-
.../commons/pipe/agent/task/PipeTaskAgent.java | 2 +-
.../pipe/agent/task/meta/PipeMetaKeeper.java | 12 +-
.../pipe/agent/task/meta/PipeStaticMeta.java | 14 +
.../config/constant/PipeExtractorConstant.java | 3 +
.../connector/client/IoTDBSyncClientManager.java | 28 +-
.../payload/thrift/request/PipeRequestType.java | 4 +-
.../thrift/request/PipeTransferFileSealReqV2.java | 13 +-
.../connector/protocol/IoTDBAirGapConnector.java | 14 +-
.../pipe/connector/protocol/IoTDBConnector.java | 10 +
.../connector/protocol/IoTDBSslSyncConnector.java | 4 +
.../options/PipeInclusionOptions.java | 43 +-
.../pipe/datastructure/pattern/TablePattern.java | 29 +-
.../pipe/datastructure/pattern/TreePattern.java | 25 +-
.../{ => resource}/PersistentResource.java | 2 +-
.../commons/pipe/datastructure/result/Result.java | 39 +-
.../pipe/datastructure/visibility/Visibility.java | 9 +-
.../visibility/VisibilityTestUtils.java | 135 +++++
.../datastructure/visibility/VisibilityUtils.java | 133 ++++
.../commons/pipe/extractor/IoTDBExtractor.java | 103 +++-
.../extractor/IoTDBNonDataRegionExtractor.java | 24 +-
.../commons/pipe/receiver/IoTDBFileReceiver.java | 49 +-
.../commons/pipe/receiver/IoTDBReceiverAgent.java | 7 +-
.../snapshot/PipeSnapshotResourceManager.java | 9 +-
.../apache/iotdb/commons/schema/node/IMNode.java | 18 +-
.../commons/schema/node/visitor/MNodeVisitor.java | 6 +-
.../schema/table/column/TsTableColumnCategory.java | 31 -
.../iotdb/commons/service/metric/enums/Metric.java | 3 +-
.../iotdb/commons/udf/access/RecordIterator.java | 9 +-
.../TableBuiltinAggregationFunction.java | 2 +
.../org/apache/iotdb/commons/utils/FileUtils.java | 49 ++
.../org/apache/iotdb/commons/utils/PathUtils.java | 2 +-
.../utils/{TestOnly.java => RetryUtils.java} | 38 +-
.../annotation/PipePluginAnnotationTest.java} | 20 +-
.../db/relational/grammar/sql/RelationalSql.g4 | 23 +-
.../thrift-commons/src/main/thrift/common.thrift | 3 +-
.../src/main/thrift/confignode.thrift | 69 +++
pom.xml | 6 +-
633 files changed, 23038 insertions(+), 7325 deletions(-)
rename integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/relational/{DatePlusOne.java => DatePlus.java} (55%)
create mode 100644 integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileTableGenerator.java
rename integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/{IoTDBRegionMigrateReliabilityITFramework.java => IoTDBRegionOperationReliabilityITFramework.java} (82%)
create mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java
create mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java
create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileWithModIT.java
copy integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/{IoTDBPipeInclusionIT.java => IoTDBPipeTableManualIT.java} (54%)
create mode 100644 integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeDoubleLivingIT.java
create mode 100644 integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeIsolationIT.java
delete mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/query/old/query/IoTDBNullValueFillTableIT.java
create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/uncorrelated/IoTDBUncorrelatedInPredicateSubqueryIT.java
create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/uncorrelated/IoTDBUncorrelatedQuantifiedComparisonIT.java
rename integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/{ => uncorrelated}/IoTDBUncorrelatedScalarSubqueryIT.java (99%)
rename integration-test/src/test/java/org/apache/iotdb/relational/it/rest/it/{IoTDBRestServiceInsertAlignedValuesIT.java => IoTDBRestServiceInsertValuesIT.java} (96%)
create mode 100644 integration-test/src/test/java/org/apache/iotdb/util/MagicUtils.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegion.java => iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TableModel.java (66%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegion.java => iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TreeModel.java (66%)
create mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/analysis/AggregateFunctionAnalysis.java
copy integration-test/src/main/java/org/apache/iotdb/itbase/category/AIClusterIT.java => iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/analysis/FunctionAnalysis.java (79%)
copy iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/{config/UDFConfigurations.java => analysis/ScalarFunctionAnalysis.java} (56%)
rename iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/parameter/{FunctionParameters.java => FunctionArguments.java} (73%)
copy iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/exception/{UDFParameterNotValidException.java => UDFArgumentNotValidException.java} (87%)
delete mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/Descriptor.java
delete mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescriptorArgument.java
delete mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/DescriptorParameterSpecification.java
delete mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/ReturnTypeSpecification.java
delete mode 100644 iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/ColumnCategory.java
copy iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeType.java => iotdb-client/session/src/main/java/org/apache/iotdb/session/util/RetryUtils.java (59%)
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/{subscription/ShowSubscriptionPlan.java => table/DescTable4InformationSchemaPlan.java} (79%)
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/{subscription/ShowSubscriptionPlan.java => table/ShowTable4InformationSchemaPlan.java} (79%)
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/{table/CommitCreateTablePlan.java => pipe/payload/PipeCreateTablePlan.java} (64%)
create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteDevicesPlan.java
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/{FetchTableResp.java => DescTable4InformationSchemaResp.java} (61%)
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/{ShowTableResp.java => ShowTable4InformationSchemaResp.java} (64%)
create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanTablePatternParseVisitor.java
rename iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/{PipeConfigPhysicalPlanPatternParseVisitor.java => PipeConfigPhysicalPlanTreePatternParseVisitor.java} (98%)
create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/ReconstructRegionProcedure.java
copy iotdb-core/{node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/plugin/env/PipeTaskRuntimeEnvironment.java => confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RegionOperationProcedure.java} (54%)
copy iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/{RegionTransitionState.java => ReconstructRegionState.java} (93%)
create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanTablePatternParseVisitorTest.java
rename iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/{PipeConfigPhysicalPlanPatternParseVisitorTest.java => PipeConfigPhysicalPlanTreePatternParseVisitorTest.java} (91%)
copy iotdb-core/{datanode/src/test/java/org/apache/iotdb/db/queryengine/exception/CpuNotEnoughExceptionTest.java => confignode/src/test/java/org/apache/iotdb/confignode/pipe/annotation/PipePluginAnnotationTest.java} (62%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/{VerifyMetadataException.java => load/LoadAnalyzeException.java} (82%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/{VerifyMetadataTypeMismatchException.java => load/LoadAnalyzeTableColumnDisorderException.java} (78%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/{VerifyMetadataTypeMismatchException.java => load/LoadAnalyzeTypeMismatchException.java} (83%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/builder/PipeTableModeTsFileBuilder.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/builder/PipeTreeModelTsFileBuilder.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/builder/PipeTsFileBuilder.java
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/{ => cacher}/LeaderCacheUtils.java (97%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTableModelTabletEventSorter.java
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/{ => sorter}/PipeTabletEventSorter.java (56%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTreeModelTabletEventSorter.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanTablePatternParseVisitor.java
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/{PipePlanPatternParseVisitor.java => PipePlanTreePatternParseVisitor.java} (98%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{queryengine/plan/relational/utils/NodeUtils.java => pipe/receiver/visitor/PipeStatementTablePatternParseVisitor.java} (55%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/{PipeStatementPatternParseVisitor.java => PipeStatementTreePatternParseVisitor.java} (97%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/{PipeStatementToBatchVisitor.java => PipeTreeStatementToBatchVisitor.java} (98%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/Slice.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/{fill/identity/IdentityLinearFill.java => function/partition/SliceCache.java} (53%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/{MergeSortInnerJoinOperator.java => MergeSortSemiJoinOperator.java} (58%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationMask.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/{CountIfAccumulator.java => CountAllAccumulator.java} (74%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/{TableScanOperator.java => aggregation/MaskedRecordIterator.java} (58%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/MarkDistinctHash.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/{SetTTLTask.java => region/ExtendRegionTask.java} (79%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/{ => region}/MigrateRegionTask.java (94%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/{DropFunctionTask.java => region/ReconstructRegionTask.java} (73%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/{SetTTLTask.java => region/RemoveRegionTask.java} (79%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SimplePlanRewriter.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneApplyColumns.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/{PruneCorrelatedJoinCorrelation.java => PruneApplyCorrelation.java} (62%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneApplySourceColumns.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/{RemoveRedundantEnforceSingleRowNode.java => RemoveUnreferencedScalarApplyNodes.java} (67%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/RemoveUnreferencedScalarSubqueries.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/TransformFilteringSemiJoinToInnerJoin.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/TransformUncorrelatedInPredicateSubqueryToSemiJoin.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/SemiJoinNode.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformQuantifiedComparisonApplyToCorrelatedJoin.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/{DescribeTable.java => CountStatement.java} (65%)
delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/Descriptor.java
delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DescriptorField.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/{SetConfiguration.java => StartRepairData.java} (85%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/{SetConfiguration.java => StopRepairData.java} (85%)
delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionDescriptorArgument.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/{sys/MergeStatement.java => metadata/region/ExtendRegionStatement.java} (74%)
rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/{ => region}/MigrateRegionStatement.java (96%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/{sys/LoadConfigurationStatement.java => metadata/region/ReconstructRegionStatement.java} (73%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/{sys/MergeStatement.java => metadata/region/RemoveRegionStatement.java} (74%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{protocol/thrift/handler/InternalServiceThriftHandlerMetrics.java => service/metrics/memory/ConsensusMemoryMetrics.java} (52%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/memory/GlobalMemoryMetrics.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/memory/QueryEngineMemoryMetrics.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/memory/SchemaEngineMemoryMetrics.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/memory/StorageEngineMemoryMetrics.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{protocol/thrift/handler/InternalServiceThriftHandlerMetrics.java => service/metrics/memory/StreamEngineMemoryMetrics.java} (51%)
delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/memory/ThresholdMemoryMetrics.java
create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementTablePatternParseVisitorTest.java
rename iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/{PipeStatementPatternParseVisitorTest.java => PipeStatementTreePatternParseVisitorTest.java} (94%)
create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanTablePatternParseVisitorTest.java
rename iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/{PipePlanPatternParseVisitorTest.java => PipePlanTreePatternParseVisitorTest.java} (94%)
copy iotdb-core/datanode/src/test/java/org/apache/iotdb/db/{queryengine/exception/CpuNotEnoughExceptionTest.java => pipe/plugin/annotation/PipePluginAnnotationTest.java} (62%)
copy iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/{SortMatcher.java => SemiJoinMatcher.java} (59%)
rename iotdb-core/{datanode/src/main/java/org/apache/iotdb/db/exception/metadata => node-commons/src/main/java/org/apache/iotdb/commons/exception}/table/ColumnNotExistsException.java (95%)
rename iotdb-core/{datanode/src/main/java/org/apache/iotdb/db/exception/metadata => node-commons/src/main/java/org/apache/iotdb/commons/exception}/table/TableAlreadyExistsException.java (95%)
rename iotdb-core/{datanode/src/main/java/org/apache/iotdb/db/exception/metadata => node-commons/src/main/java/org/apache/iotdb/commons/exception}/table/TableNotExistsException.java (95%)
rename iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/{ => resource}/PersistentResource.java (95%)
copy library-udf/src/main/java/org/apache/iotdb/library/match/model/SectionNext.java => iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/result/Result.java (59%)
copy integration-test/src/main/java/org/apache/iotdb/itbase/category/MultiClusterIT2SubscriptionRegressionConsumer.java => iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/Visibility.java (86%)
create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityTestUtils.java
create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
copy iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/{TestOnly.java => RetryUtils.java} (60%)
copy iotdb-core/{datanode/src/test/java/org/apache/iotdb/db/queryengine/exception/CpuNotEnoughExceptionTest.java => node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/annotation/PipePluginAnnotationTest.java} (62%)
(iotdb) 01/03: save
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udtf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 87fc892d0725e7e70b101ce47a86520e8fd01611
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Mon Jan 20 19:05:23 2025 +0800
save
---
.../relational/table/argument/DescribedSchema.java | 20 ++++--------
.../relational/table/argument/TableArgument.java | 11 -------
.../apache/iotdb/udf/api/type/ColumnCategory.java | 37 ----------------------
.../function/table/ExcludeColumnFunction.java | 3 +-
.../execution/function/table/HOPTableFunction.java | 11 ++++---
.../execution/function/table/SplitFunction.java | 4 +--
.../relational/analyzer/StatementAnalyzer.java | 11 ++-----
.../plan/relational/sql/parser/AstBuilder.java | 21 +++++++++++-
.../schema/table/column/TsTableColumnCategory.java | 31 ------------------
.../db/relational/grammar/sql/RelationalSql.g4 | 7 +++-
10 files changed, 42 insertions(+), 114 deletions(-)
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescribedSchema.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescribedSchema.java
index 95411b7513a..aecea22067f 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescribedSchema.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescribedSchema.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.udf.api.relational.table.argument;
-import org.apache.iotdb.udf.api.type.ColumnCategory;
import org.apache.iotdb.udf.api.type.Type;
import java.util.ArrayList;
@@ -50,13 +49,13 @@ public class DescribedSchema {
public static class Builder {
private final List<Field> fields = new ArrayList<>();
- public Builder addField(String name, Type type, ColumnCategory category) {
- fields.add(new Field(name, type, category));
+ public Builder addField(String name, Type type) {
+ fields.add(new Field(name, type));
return this;
}
- public Builder addField(Optional<String> name, Type type, ColumnCategory category) {
- fields.add(new Field(name, type, category));
+ public Builder addField(Optional<String> name, Type type) {
+ fields.add(new Field(name, type));
return this;
}
@@ -68,18 +67,15 @@ public class DescribedSchema {
public static class Field {
private final Optional<String> name;
private final Type type;
- private final ColumnCategory category;
- public Field(String name, Type type, ColumnCategory category) {
+ public Field(String name, Type type) {
this.name = Optional.ofNullable(name);
this.type = type;
- this.category = category;
}
- public Field(Optional<String> name, Type type, ColumnCategory category) {
+ public Field(Optional<String> name, Type type) {
this.name = name;
this.type = type;
- this.category = category;
}
public Optional<String> getName() {
@@ -89,9 +85,5 @@ public class DescribedSchema {
public Type getType() {
return type;
}
-
- public ColumnCategory getCategory() {
- return category;
- }
}
}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/TableArgument.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/TableArgument.java
index d989f526694..4160caa52e6 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/TableArgument.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/TableArgument.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.udf.api.relational.table.argument;
-import org.apache.iotdb.udf.api.type.ColumnCategory;
import org.apache.iotdb.udf.api.type.Type;
import java.util.List;
@@ -30,25 +29,19 @@ import static java.util.Objects.requireNonNull;
public class TableArgument implements Argument {
private final List<Optional<String>> fieldNames;
private final List<Type> fieldTypes;
- private final List<ColumnCategory> fieldCategories;
private final List<String> partitionBy;
private final List<String> orderBy;
public TableArgument(
List<Optional<String>> fieldNames,
List<Type> fieldTypes,
- List<ColumnCategory> fieldCategories,
List<String> partitionBy,
List<String> orderBy) {
this.fieldNames = requireNonNull(fieldNames, "fieldNames is null");
this.fieldTypes = requireNonNull(fieldTypes, "fieldTypes is null");
- this.fieldCategories = requireNonNull(fieldCategories, "fieldCategories is null");
if (fieldNames.size() != fieldTypes.size()) {
throw new IllegalArgumentException("fieldNames and fieldTypes must have the same size");
}
- if (fieldNames.size() != fieldCategories.size()) {
- throw new IllegalArgumentException("fieldNames and fieldCategories must have the same size");
- }
this.partitionBy = requireNonNull(partitionBy, "partitionBy is null");
this.orderBy = requireNonNull(orderBy, "orderBy is null");
}
@@ -61,10 +54,6 @@ public class TableArgument implements Argument {
return fieldTypes;
}
- public List<ColumnCategory> getFieldCategories() {
- return fieldCategories;
- }
-
public List<String> getPartitionBy() {
return partitionBy;
}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/ColumnCategory.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/ColumnCategory.java
deleted file mode 100644
index 8ac065bf942..00000000000
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/ColumnCategory.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.udf.api.type;
-
-public enum ColumnCategory {
- TAG((byte) 0),
- ATTRIBUTE((byte) 1),
- TIME((byte) 2),
- FIELD((byte) 3);
-
- private final byte category;
-
- ColumnCategory(byte category) {
- this.category = category;
- }
-
- byte getValue() {
- return category;
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
index 5ba833ed7a2..2b3998365a5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
@@ -68,8 +68,7 @@ public class ExcludeColumnFunction implements TableFunction {
requiredColumns.add(i);
schemaBuilder.addField(
fieldName,
- tableArgument.getFieldTypes().get(i),
- tableArgument.getFieldCategories().get(i));
+ tableArgument.getFieldTypes().get(i));
}
}
return TableFunctionAnalysis.builder()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
index ce4021a0771..6bf8df15f04 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
@@ -32,7 +32,6 @@ import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProc
import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
-import org.apache.iotdb.udf.api.type.ColumnCategory;
import org.apache.iotdb.udf.api.type.Type;
import org.apache.tsfile.block.column.ColumnBuilder;
@@ -85,6 +84,7 @@ public class HOPTableFunction implements TableFunction {
return requiredIndex;
}
+ // TODO: ImmutableMap
@Override
public TableFunctionAnalysis analyze(Map<String, Argument> arguments) {
TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
@@ -96,10 +96,11 @@ public class HOPTableFunction implements TableFunction {
}
DescribedSchema properColumnSchema =
new DescribedSchema.Builder()
- .addField("window_start", Type.TIMESTAMP, ColumnCategory.FIELD)
- .addField("window_end", Type.TIMESTAMP, ColumnCategory.FIELD)
+ .addField("window_start", Type.TIMESTAMP)
+ .addField("window_end", Type.TIMESTAMP)
.build();
+ // outputColumnSchema
return TableFunctionAnalysis.builder()
.properColumnSchema(properColumnSchema)
.requiredColumns(
@@ -122,8 +123,8 @@ public class HOPTableFunction implements TableFunction {
public TableFunctionDataProcessor getDataProcessor() {
return new HOPDataProcessor(
(Long) ((ScalarArgument) arguments.get(START_PARAMETER_NAME)).getValue(),
- (Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue() * 1000,
- (Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue() * 1000,
+ (Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue(),
+ (Long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue(),
requiredIndex);
}
};
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
index 263cfa79d7f..76880270f37 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
@@ -29,7 +29,6 @@ import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionLeafProcessor;
import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
-import org.apache.iotdb.udf.api.type.ColumnCategory;
import org.apache.iotdb.udf.api.type.Type;
import org.apache.tsfile.block.column.ColumnBuilder;
@@ -57,8 +56,7 @@ public class SplitFunction implements TableFunction {
@Override
public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException {
- DescribedSchema schema =
- DescribedSchema.builder().addField("output", Type.STRING, ColumnCategory.FIELD).build();
+ DescribedSchema schema = DescribedSchema.builder().addField("output", Type.STRING).build();
return TableFunctionAnalysis.builder().properColumnSchema(schema).build();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index 9528f7e2e2f..bbdb55815ac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@ -174,7 +174,6 @@ import org.apache.iotdb.udf.api.relational.table.specification.DescriptorParamet
import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
-import org.apache.iotdb.udf.api.type.ColumnCategory;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
@@ -3196,7 +3195,7 @@ public class StatementAnalyzer {
Field.newUnqualified(
f.getName(),
UDFDataTypeTransformer.transformUDFDataTypeToReadType(f.getType()),
- TsTableColumnCategory.fromUdfColumnType(f.getCategory())))
+ TsTableColumnCategory.FIELD))
.forEach(fields::add));
// next, columns derived from table arguments, in order of argument declarations
@@ -3394,7 +3393,6 @@ public class StatementAnalyzer {
Optional<Scope> scope) {
List<Optional<String>> fieldNames;
List<org.apache.iotdb.udf.api.type.Type> fieldTypes;
- List<ColumnCategory> fieldCategories;
List<String> partitionBy = Collections.emptyList();
List<String> orderBy = Collections.emptyList();
@@ -3418,11 +3416,6 @@ public class StatementAnalyzer {
.map(Field::getType)
.map(UDFDataTypeTransformer::transformReadTypeToUDFDataType)
.collect(toImmutableList());
- fieldCategories =
- fields.stream()
- .map(Field::getColumnCategory)
- .map(TsTableColumnCategory::toUdfColumnCategory)
- .collect(toImmutableList());
// analyze PARTITION BY
if (tableArgument.getPartitionBy().isPresent()) {
@@ -3522,7 +3515,7 @@ public class StatementAnalyzer {
analysisBuilder.withPassThroughColumns(argumentSpecification.isPassThroughColumns());
return new ArgumentAnalysis(
- new TableArgument(fieldNames, fieldTypes, fieldCategories, partitionBy, orderBy),
+ new TableArgument(fieldNames, fieldTypes, partitionBy, orderBy),
Optional.of(analysisBuilder.build()));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index b4f40236b6c..482c6bf5e93 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -243,6 +243,7 @@ import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.GroupingSe
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.QualifiedName.mapIdentifier;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument.descriptorArgument;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument.nullDescriptorArgument;
+import static org.apache.iotdb.db.utils.TimestampPrecisionUtils.currPrecision;
import static org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_AGGREGATION;
import static org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_BY_AGGREGATION;
import static org.apache.iotdb.db.utils.constant.SqlConstant.LAST_AGGREGATION;
@@ -1834,7 +1835,7 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
} else if (context.descriptorArgument() != null) {
value = visit(context.descriptorArgument());
} else {
- value = visit(context.expression());
+ value = visit(context.scalarArgument());
}
return new TableFunctionArgument(getLocation(context), name, value);
@@ -1931,6 +1932,24 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
visitIfPresent(context.type(), DataType.class));
}
+ @Override
+ public Node visitScalarArgument(RelationalSqlParser.ScalarArgumentContext ctx) {
+ if (ctx.expression() != null) {
+ return visit(ctx.expression());
+ } else {
+ TimeDuration timeDuration = DateTimeUtils.constructTimeDuration(ctx.timeDuration().getText());
+
+ if (timeDuration.monthDuration != 0 && timeDuration.nonMonthDuration != 0) {
+ throw new SemanticException(
+ "Simultaneous setting of monthly and non-monthly intervals is not supported.");
+ }
+
+ return new LongLiteral(
+ getLocation(ctx.timeDuration()),
+ String.valueOf(timeDuration.getTotalDuration(currPrecision)));
+ }
+ }
+
// ********************* predicates *******************
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java
index 4ff0c6663b9..9f6dcd6019c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java
@@ -102,37 +102,6 @@ public enum TsTableColumnCategory {
}
}
- public org.apache.iotdb.udf.api.type.ColumnCategory toUdfColumnCategory() {
- switch (this) {
- case TAG:
- return org.apache.iotdb.udf.api.type.ColumnCategory.TAG;
- case ATTRIBUTE:
- return org.apache.iotdb.udf.api.type.ColumnCategory.ATTRIBUTE;
- case FIELD:
- return org.apache.iotdb.udf.api.type.ColumnCategory.FIELD;
- case TIME:
- return org.apache.iotdb.udf.api.type.ColumnCategory.TIME;
- default:
- throw new IllegalArgumentException("Unsupported column type in UDF: " + this);
- }
- }
-
- public static TsTableColumnCategory fromUdfColumnType(
- org.apache.iotdb.udf.api.type.ColumnCategory columnType) {
- switch (columnType) {
- case FIELD:
- return FIELD;
- case TAG:
- return TAG;
- case ATTRIBUTE:
- return ATTRIBUTE;
- case TIME:
- return TIME;
- default:
- throw new IllegalArgumentException("Unknown column type: " + columnType);
- }
- }
-
public byte getCategory() {
return category;
}
diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
index e53900465a5..4f0ae953dac 100644
--- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
+++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
@@ -735,7 +735,7 @@ tableFunctionCall
;
tableFunctionArgument
- : (identifier '=>')? (tableArgument | descriptorArgument | expression) // descriptor before expression to avoid parsing descriptor as a function call
+ : (identifier '=>')? (tableArgument | descriptorArgument | scalarArgument) // descriptor before expression to avoid parsing descriptor as a function call
;
tableArgument
@@ -759,6 +759,11 @@ descriptorField
: identifier type?
;
+scalarArgument
+ : expression
+ | timeDuration
+ ;
+
copartitionTables
: '(' qualifiedName ',' qualifiedName (',' qualifiedName)* ')'
;
(iotdb) 03/03: adjust api
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udtf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b9dd299cbe8799ced4f0314dd0c3f682e552f323
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Wed Feb 5 18:06:09 2025 +0800
adjust api
---
.../iotdb/udf/api/relational/TableFunction.java | 51 ++++++-
.../relational/table/TableFunctionAnalysis.java | 4 +-
.../api/relational/table/argument/Descriptor.java | 95 -------------
.../table/argument/DescriptorArgument.java | 57 --------
.../processor/TableFunctionDataProcessor.java | 32 +++--
.../processor/TableFunctionLeafProcessor.java | 5 +
.../DescriptorParameterSpecification.java | 57 --------
.../specification/ReturnTypeSpecification.java | 68 ---------
.../specification/TableParameterSpecification.java | 9 --
.../function/table/ExcludeColumnFunction.java | 27 +---
.../execution/function/table/HOPTableFunction.java | 47 ++-----
.../execution/function/table/SplitFunction.java | 2 +-
.../process/function/PartitionRecognizer.java | 124 +++--------------
.../process/function/TableFunctionOperator.java | 73 ++++++++--
.../process/function/partition/PartitionState.java | 22 ++-
.../operator/process/function/partition/Slice.java | 153 +++++++++++++++++++++
.../process/function/partition/SliceCache.java | 55 ++++++++
.../plan/planner/TableOperatorGenerator.java | 7 +
.../plan/planner/plan/node/PlanGraphPrinter.java | 22 +--
.../relational/analyzer/StatementAnalyzer.java | 108 ++++-----------
.../plan/relational/sql/ast/AstVisitor.java | 13 --
.../plan/relational/sql/ast/Descriptor.java | 80 -----------
.../plan/relational/sql/ast/DescriptorField.java | 88 ------------
.../relational/sql/ast/TableFunctionArgument.java | 5 +-
.../sql/ast/TableFunctionDescriptorArgument.java | 87 ------------
.../plan/relational/sql/parser/AstBuilder.java | 24 ----
.../plan/relational/sql/util/SqlFormatter.java | 23 ----
.../process/tvf/TableFunctionOperatorTest.java | 57 ++++----
.../db/relational/grammar/sql/RelationalSql.g4 | 11 +-
29 files changed, 466 insertions(+), 940 deletions(-)
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/TableFunction.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/TableFunction.java
index fb0e89f9557..cf74cbd748f 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/TableFunction.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/TableFunction.java
@@ -23,16 +23,65 @@ import org.apache.iotdb.udf.api.exception.UDFException;
import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
import java.util.List;
import java.util.Map;
public interface TableFunction {
- List<ParameterSpecification> getArgumentsSpecification();
+ /**
+ * This method is used to define the specification of the arguments required by the table
+ * function. Each argument is described using a {@link ParameterSpecification} object, which
+ * encapsulates details such as the argument name, whether it is required, its default value (if
+ * any), etc.
+ *
+ * <p>The {@link ParameterSpecification} class is abstract and has two concrete implementations:
+ *
+ * <ul>
+ * <li>{@link TableParameterSpecification}: Used for parameters specific to table functions.
+ * <li>{@link ScalarParameterSpecification}: Used for parameters specific to scalar functions.
+ * </ul>
+ *
+ * @return a list of {@link ParameterSpecification} objects describing the function's arguments.
+ * The list should include all parameters that the table function expects, along with their
+ * constraints and default values (if applicable).
+ */
+ List<ParameterSpecification> getArgumentsSpecifications();
+ /**
+ * This method is responsible for analyzing the provided arguments and constructing a {@link
+ * TableFunctionAnalysis} object in runtime. During analysis, the method should:
+ *
+ * <ul>
+ * <li>Extract values from scalar arguments (instances of {@link ScalarArgument}) or extract
+ * table descriptions from table arguments (instances of {@link TableArgument}).
+ * <li>Construct a {@link TableFunctionAnalysis} object that contains:
+ * <ul>
+ * <li>A description of proper columns.
+ * <li>A map indicating which columns from the input table arguments are required for the
+ * function to execute.
+ * </ul>
+ * </ul>
+ *
+ * @param arguments a map of argument names to their corresponding {@link Argument} values. The
+ * keys should match the parameter names defined in {@link #getArgumentsSpecifications()}.
+ * @throws UDFException if any argument is invalid or if an error occurs during analysis
+ * @return a {@link TableFunctionAnalysis} object containing the analysis result
+ */
TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException;
+ /**
+ * This method is used to obtain a {@link TableFunctionProcessorProvider} that will be responsible
+ * for creating processors to handle the transformation of input data into output table. The
+ * provider is initialized with the validated arguments.
+ *
+ * @param arguments a map of argument names to their corresponding {@link Argument} values
+ * @return a {@link TableFunctionProcessorProvider} for creating processors
+ */
TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments);
}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionAnalysis.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionAnalysis.java
index d535518cbbd..b33eb8c8ac7 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionAnalysis.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionAnalysis.java
@@ -35,8 +35,8 @@ import static java.util.Objects.requireNonNull;
public class TableFunctionAnalysis {
/**
- * The `returnedSchema` field is used to inform the Analyzer of the proper columns returned by the
- * Table Function, that is, the columns produced by the function.
+ * The `properColumnSchema` field is used to inform the Analyzer of the proper columns returned by
+ * the Table Function, that is, the columns produced by the function.
*/
private final Optional<DescribedSchema> properColumnSchema;
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/Descriptor.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/Descriptor.java
deleted file mode 100644
index ae9624f8f0b..00000000000
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/Descriptor.java
+++ /dev/null
@@ -1,95 +0,0 @@
-package org.apache.iotdb.udf.api.relational.table.argument;
-
-import org.apache.iotdb.udf.api.type.Type;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-
-import static java.util.Objects.requireNonNull;
-
-public class Descriptor {
- private final List<Field> fields;
-
- public Descriptor(List<Field> fields) {
- requireNonNull(fields, "fields is null");
- if (fields.isEmpty()) {
- throw new IllegalArgumentException("descriptor has no fields");
- }
- this.fields = fields;
- }
-
- public static Descriptor descriptor(List<String> names, List<Type> types) {
- requireNonNull(names, "names is null");
- requireNonNull(types, "types is null");
- if (names.size() != types.size()) {
- throw new IllegalArgumentException("names and types lists do not match");
- }
- List<Field> fields = new ArrayList<>();
- for (int i = 0; i < names.size(); i++) {
- fields.add(new Field(names.get(i), Optional.of(types.get(i))));
- }
- return new Descriptor(fields);
- }
-
- public List<Field> getFields() {
- return fields;
- }
-
- public boolean isTyped() {
- return fields.stream().allMatch(field -> field.type.isPresent());
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- Descriptor that = (Descriptor) o;
- return fields.equals(that.fields);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(fields);
- }
-
- public static class Field {
- private final String name;
- private final Optional<Type> type;
-
- public Field(String name, Optional<Type> type) {
- this.name = name;
- this.type = type;
- }
-
- public String getName() {
- return name;
- }
-
- public Optional<Type> getType() {
- return type;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- Field field = (Field) o;
- return name.equals(field.name) && type.equals(field.type);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(name, type);
- }
- }
-}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescriptorArgument.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescriptorArgument.java
deleted file mode 100644
index 30127511df1..00000000000
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/argument/DescriptorArgument.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.udf.api.relational.table.argument;
-
-import java.util.Objects;
-import java.util.Optional;
-
-import static java.util.Objects.requireNonNull;
-
-public class DescriptorArgument implements Argument {
- public static final DescriptorArgument NULL_DESCRIPTOR = new DescriptorArgument(Optional.empty());
-
- // it is optional because it is possible that the descriptor is not present(null descriptor)
- private final Optional<Descriptor> descriptor;
-
- public DescriptorArgument(Optional<Descriptor> descriptor) {
- this.descriptor = requireNonNull(descriptor, "descriptor is null");
- }
-
- public Optional<Descriptor> getDescriptor() {
- return descriptor;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- DescriptorArgument that = (DescriptorArgument) o;
- return descriptor.equals(that.descriptor);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(descriptor);
- }
-}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionDataProcessor.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionDataProcessor.java
index d238d0abcb1..c3cd5483b64 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionDataProcessor.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionDataProcessor.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.udf.api.relational.table.processor;
-import org.apache.iotdb.udf.api.relational.TableFunction;
import org.apache.iotdb.udf.api.relational.access.Record;
import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
@@ -30,23 +29,36 @@ import java.util.List;
/** Each instance of TableFunctionDataProcessor processes one partition of data. */
public interface TableFunctionDataProcessor {
+ default void beforeStart() {
+ // do nothing
+ }
+
/**
* This method processes a portion of data. It is called multiple times until the partition is
* fully processed.
*
* @param input {@link Record} including a portion of one partition for each table function's
- * input table. Input is ordered according to the corresponding argument specifications in
- * {@link TableFunction}. A page for an argument consists of columns requested during analysis
- * (see {@link TableFunctionAnalysis#getRequiredColumns}).
- * @param columnBuilders a list of {@link ColumnBuilder} for each column in the output table.
+ * input table. A Record consists of columns requested during analysis (see {@link
+ * TableFunctionAnalysis#getRequiredColumns}).
+ * @param properColumnBuilders A list of {@link ColumnBuilder} for each column in the output
+ * table.
+ * @param passThroughIndexBuilder A {@link ColumnBuilder} for pass through columns. Index is
+ * started from 0 of the whole partition.
*/
- void process(Record input, List<ColumnBuilder> columnBuilders);
+ void process(
+ Record input,
+ List<ColumnBuilder> properColumnBuilders,
+ ColumnBuilder passThroughIndexBuilder);
/**
- * This method is called after all data is processed. It is used to finalize the output table. All
- * remaining data should be written to the columnBuilders.
+ * This method is called after all data is processed. It is used to finalize the output table and
+ * close resource. All remaining data should be written to the columnBuilders.
*
- * @param columnBuilders a list of {@link ColumnBuilder} for each column in the output table.
+ * @param columnBuilders A list of {@link ColumnBuilder} for each column in the output table.
+ * @param passThroughIndexBuilder A {@link ColumnBuilder} for pass through columns. Index is
+ * started from 0 of the whole partition.
*/
- void finish(List<ColumnBuilder> columnBuilders);
+ default void finish(List<ColumnBuilder> columnBuilders, ColumnBuilder passThroughIndexBuilder) {
+ // do nothing
+ }
}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionLeafProcessor.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionLeafProcessor.java
index a8a49af3fa8..41750fc3b53 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionLeafProcessor.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/processor/TableFunctionLeafProcessor.java
@@ -24,6 +24,11 @@ import org.apache.tsfile.block.column.ColumnBuilder;
import java.util.List;
public interface TableFunctionLeafProcessor {
+
+ default void beforeStart() {
+ // do nothing
+ }
+
/**
* This method processes a portion of data. It is called multiple times until the processor is
* fully processed.
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/DescriptorParameterSpecification.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/DescriptorParameterSpecification.java
deleted file mode 100644
index 06a0fcdee62..00000000000
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/DescriptorParameterSpecification.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.udf.api.relational.table.specification;
-
-import org.apache.iotdb.udf.api.relational.table.argument.Descriptor;
-
-import java.util.Optional;
-
-public class DescriptorParameterSpecification extends ParameterSpecification {
- private DescriptorParameterSpecification(String name, boolean required, Descriptor defaultValue) {
- super(name, required, Optional.ofNullable(defaultValue));
- }
-
- public static Builder builder() {
- return new Builder();
- }
-
- public static final class Builder {
- private String name;
- private boolean required = true;
- private Descriptor defaultValue;
-
- private Builder() {}
-
- public Builder name(String name) {
- this.name = name;
- return this;
- }
-
- public Builder defaultValue(Descriptor defaultValue) {
- this.required = false;
- this.defaultValue = defaultValue;
- return this;
- }
-
- public DescriptorParameterSpecification build() {
- return new DescriptorParameterSpecification(name, required, defaultValue);
- }
- }
-}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/ReturnTypeSpecification.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/ReturnTypeSpecification.java
deleted file mode 100644
index 03e4f7fa7da..00000000000
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/ReturnTypeSpecification.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.udf.api.relational.table.specification;
-
-import org.apache.iotdb.udf.api.relational.table.argument.Descriptor;
-
-import static java.util.Objects.requireNonNull;
-
-/**
- * The return type declaration refers to the proper columns of the table function. These are the
- * columns produced by the table function as opposed to the columns of input relations passed
- * through by the table function.
- */
-public abstract class ReturnTypeSpecification {
- /**
- * The proper columns of the table function are not known at function declaration time. They must
- * be determined at query analysis time based on the actual call arguments.
- */
- public static class GenericTable extends ReturnTypeSpecification {
- public static final GenericTable GENERIC_TABLE = new GenericTable();
-
- private GenericTable() {}
- }
-
- /** The table function has no proper columns. */
- public static class OnlyPassThrough extends ReturnTypeSpecification {
- public static final OnlyPassThrough ONLY_PASS_THROUGH = new OnlyPassThrough();
-
- private OnlyPassThrough() {}
- }
-
- /**
- * The proper columns of the table function are known at function declaration time. They do not
- * depend on the actual call arguments.
- */
- public static class DescribedTable extends ReturnTypeSpecification {
- private final Descriptor descriptor;
-
- public DescribedTable(Descriptor descriptor) {
- requireNonNull(descriptor, "descriptor is null");
- if (!descriptor.isTyped()) {
- throw new IllegalArgumentException("field types not specified");
- }
- this.descriptor = descriptor;
- }
-
- public Descriptor getDescriptor() {
- return descriptor;
- }
- }
-}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java
index 7348ec54355..eabd6d61a5b 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java
@@ -32,10 +32,6 @@ public class TableParameterSpecification extends ParameterSpecification {
String name, boolean rowSemantics, boolean pruneWhenEmpty, boolean passThroughColumns) {
// table arguments are always required
super(name, true, Optional.empty());
- // if (rowSemantics && !pruneWhenEmpty) {
- // throw new IllegalArgumentException(
- // "Cannot set the KEEP WHEN EMPTY property for a table argument with row semantics");
- // }
this.rowSemantics = rowSemantics;
this.pruneWhenEmpty = pruneWhenEmpty;
this.passThroughColumns = passThroughColumns;
@@ -80,11 +76,6 @@ public class TableParameterSpecification extends ParameterSpecification {
return this;
}
- public Builder pruneWhenEmpty() {
- this.pruneWhenEmpty = true;
- return this;
- }
-
public Builder keepWhenEmpty() {
this.pruneWhenEmpty = false;
return this;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
index e01428e4736..63b727e836c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.queryengine.execution.function.table;
import org.apache.iotdb.udf.api.exception.UDFException;
import org.apache.iotdb.udf.api.relational.TableFunction;
-import org.apache.iotdb.udf.api.relational.access.Record;
import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
import org.apache.iotdb.udf.api.relational.table.argument.Argument;
@@ -35,7 +34,6 @@ import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpe
import org.apache.iotdb.udf.api.type.Type;
import com.google.common.collect.ImmutableList;
-import org.apache.tsfile.block.column.ColumnBuilder;
import java.util.Arrays;
import java.util.List;
@@ -47,7 +45,7 @@ public class ExcludeColumnFunction implements TableFunction {
private final String COL_PARAM = "EXCLUDE";
@Override
- public List<ParameterSpecification> getArgumentsSpecification() {
+ public List<ParameterSpecification> getArgumentsSpecifications() {
return Arrays.asList(
TableParameterSpecification.builder().name(TBL_PARAM).rowSemantics().build(),
ScalarParameterSpecification.builder().name(COL_PARAM).type(Type.STRING).build());
@@ -56,9 +54,6 @@ public class ExcludeColumnFunction implements TableFunction {
@Override
public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException {
TableArgument tableArgument = (TableArgument) arguments.get(TBL_PARAM);
- if (tableArgument == null) {
- throw new UDFException("Table argument is missing");
- }
String excludeColumn = (String) ((ScalarArgument) arguments.get(COL_PARAM)).getValue();
ImmutableList.Builder<Integer> requiredColumns = ImmutableList.builder();
DescribedSchema.Builder schemaBuilder = DescribedSchema.builder();
@@ -80,22 +75,14 @@ public class ExcludeColumnFunction implements TableFunction {
return new TableFunctionProcessorProvider() {
@Override
public TableFunctionDataProcessor getDataProcessor() {
- return new TableFunctionDataProcessor() {
- @Override
- public void process(Record input, List<ColumnBuilder> columnBuilders) {
- for (int i = 0; i < input.size(); i++) {
- if (input.isNull(i)) {
- columnBuilders.get(i).appendNull();
- } else {
- columnBuilders.get(i).writeObject(input.getObject(i));
- }
+ return (input, properColumnBuilders, passThroughIndexBuilder) -> {
+ for (int i = 0; i < input.size(); i++) {
+ if (input.isNull(i)) {
+ properColumnBuilders.get(i).appendNull();
+ } else {
+ properColumnBuilders.get(i).writeObject(input.getObject(i));
}
}
-
- @Override
- public void finish(List<ColumnBuilder> columnBuilders) {
- // do nothing
- }
};
}
};
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
index 4da85a53f12..470e4aa987b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
@@ -41,8 +41,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
public class HOPTableFunction implements TableFunction {
@@ -53,7 +51,7 @@ public class HOPTableFunction implements TableFunction {
private static final String START_PARAMETER_NAME = "START";
@Override
- public List<ParameterSpecification> getArgumentsSpecification() {
+ public List<ParameterSpecification> getArgumentsSpecifications() {
return Arrays.asList(
TableParameterSpecification.builder()
.name(DATA_PARAMETER_NAME)
@@ -85,7 +83,6 @@ public class HOPTableFunction implements TableFunction {
return requiredIndex;
}
- // TODO: ImmutableMap
@Override
public TableFunctionAnalysis analyze(Map<String, Argument> arguments) {
TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
@@ -104,29 +101,19 @@ public class HOPTableFunction implements TableFunction {
// outputColumnSchema
return TableFunctionAnalysis.builder()
.properColumnSchema(properColumnSchema)
- .requiredColumns(
- DATA_PARAMETER_NAME,
- IntStream.range(0, tableArgument.getFieldTypes().size())
- .boxed()
- .collect(Collectors.toList()))
+ .requiredColumns(DATA_PARAMETER_NAME, Collections.singletonList(requiredIndex))
.build();
}
@Override
public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
- TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
- String expectedFieldName =
- (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
- int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName);
-
return new TableFunctionProcessorProvider() {
@Override
public TableFunctionDataProcessor getDataProcessor() {
return new HOPDataProcessor(
(Long) ((ScalarArgument) arguments.get(START_PARAMETER_NAME)).getValue(),
(Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue(),
- (Long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue(),
- requiredIndex);
+ (Long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue());
}
};
}
@@ -135,19 +122,21 @@ public class HOPTableFunction implements TableFunction {
private final long slide;
private final long size;
- private final int timeColumnIndex;
private long curTime;
+ private long curIndex = 0;
- public HOPDataProcessor(long startTime, long slide, long size, int timeColumnIndex) {
+ public HOPDataProcessor(long startTime, long slide, long size) {
this.slide = slide;
this.size = size;
this.curTime = startTime;
- this.timeColumnIndex = timeColumnIndex;
}
@Override
- public void process(Record input, List<ColumnBuilder> columnBuilders) {
- long timeValue = input.getLong(timeColumnIndex);
+ public void process(
+ Record input,
+ List<ColumnBuilder> properColumnBuilders,
+ ColumnBuilder passThroughIndexBuilder) {
+ long timeValue = input.getLong(0);
if (curTime == Long.MIN_VALUE) {
curTime = timeValue;
}
@@ -158,20 +147,12 @@ public class HOPTableFunction implements TableFunction {
}
long slideTime = curTime;
while (slideTime <= timeValue && slideTime + size > timeValue) {
- for (int i = 0; i < input.size(); i++) {
- if (input.isNull(i)) {
- columnBuilders.get(i + 2).appendNull();
- } else {
- columnBuilders.get(i + 2).writeObject(input.getObject(i));
- }
- }
- columnBuilders.get(0).writeLong(slideTime);
- columnBuilders.get(1).writeLong(slideTime + size);
+ properColumnBuilders.get(0).writeLong(slideTime);
+ properColumnBuilders.get(1).writeLong(slideTime + size);
+ passThroughIndexBuilder.writeLong(curIndex);
slideTime += slide;
}
+ curIndex++;
}
-
- @Override
- public void finish(List<ColumnBuilder> columnBuilders) {}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
index 76880270f37..d000940678a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/SplitFunction.java
@@ -44,7 +44,7 @@ public class SplitFunction implements TableFunction {
private final String SPLIT_PARAMETER_NAME = "SPLIT";
@Override
- public List<ParameterSpecification> getArgumentsSpecification() {
+ public List<ParameterSpecification> getArgumentsSpecifications() {
return Arrays.asList(
ScalarParameterSpecification.builder().name(INPUT_PARAMETER_NAME).type(Type.STRING).build(),
ScalarParameterSpecification.builder()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/PartitionRecognizer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/PartitionRecognizer.java
index 09be3664354..6068aeadc39 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/PartitionRecognizer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/PartitionRecognizer.java
@@ -21,21 +21,14 @@ package org.apache.iotdb.db.queryengine.execution.operator.process.function;
import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.PartitionState;
-import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.Slice;
import org.apache.iotdb.udf.api.type.Type;
-import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.utils.Binary;
-import org.apache.tsfile.utils.DateUtils;
-import java.time.LocalDate;
import java.util.ArrayList;
-import java.util.Iterator;
import java.util.List;
-import java.util.NoSuchElementException;
import java.util.Objects;
public class PartitionRecognizer {
@@ -43,7 +36,7 @@ public class PartitionRecognizer {
private final List<Integer> partitionChannels;
private final List<Object> partitionValues;
private final int[] requiredChannels;
- private final List<Type> outputDataTypes;
+ private final List<Type> inputDataTypes;
private TsBlock currentTsBlock = null;
private boolean noMoreData = false;
private int currentIndex = 0;
@@ -52,14 +45,14 @@ public class PartitionRecognizer {
public PartitionRecognizer(
List<Integer> partitionChannels,
List<Integer> requiredChannels,
- List<TSDataType> outputDataTypes) {
+ List<TSDataType> inputDataTypes) {
this.partitionChannels = partitionChannels;
this.partitionValues = new ArrayList<>(partitionChannels.size());
for (int i = 0; i < partitionChannels.size(); i++) {
partitionValues.add(null);
}
this.requiredChannels = requiredChannels.stream().mapToInt(i -> i).toArray();
- this.outputDataTypes = UDFDataTypeTransformer.transformToUDFDataTypeList(outputDataTypes);
+ this.inputDataTypes = UDFDataTypeTransformer.transformToUDFDataTypeList(inputDataTypes);
}
// TsBlock is sorted by partition columns already
@@ -76,7 +69,7 @@ public class PartitionRecognizer {
noMoreData = true;
}
- public PartitionState getState() {
+ public PartitionState nextState() {
updateState();
return state;
}
@@ -109,9 +102,9 @@ public class PartitionRecognizer {
return PartitionState.INIT_STATE;
}
int endPartitionIndex = findNextDifferentRowIndex();
- Iterator<Record> recordIterator = getRecordIterator(currentIndex, endPartitionIndex);
+ Slice slice = getSlice(currentIndex, endPartitionIndex);
currentIndex = endPartitionIndex;
- return PartitionState.newPartitionState(recordIterator);
+ return PartitionState.newPartitionState(slice);
}
private PartitionState handleNewPartitionState() {
@@ -119,9 +112,9 @@ public class PartitionRecognizer {
return PartitionState.NEED_MORE_DATA_STATE;
} else {
int endPartitionIndex = findNextDifferentRowIndex();
- Iterator<Record> recordIterator = getRecordIterator(currentIndex, endPartitionIndex);
+ Slice slice = getSlice(currentIndex, endPartitionIndex);
currentIndex = endPartitionIndex;
- return PartitionState.newPartitionState(recordIterator);
+ return PartitionState.newPartitionState(slice);
}
}
@@ -133,15 +126,15 @@ public class PartitionRecognizer {
}
int endPartitionIndex = findNextDifferentRowIndex();
if (endPartitionIndex != 0) {
- Iterator<Record> recordIterator = getRecordIterator(currentIndex, endPartitionIndex);
+ Slice slice = getSlice(currentIndex, endPartitionIndex);
currentIndex = endPartitionIndex;
- return PartitionState.iteratingState(recordIterator);
+ return PartitionState.iteratingState(slice);
} else {
currentIndex = endPartitionIndex;
endPartitionIndex = findNextDifferentRowIndex();
- Iterator<Record> recordIterator = getRecordIterator(currentIndex, endPartitionIndex);
+ Slice slice = getSlice(currentIndex, endPartitionIndex);
currentIndex = endPartitionIndex;
- return PartitionState.newPartitionState(recordIterator);
+ return PartitionState.newPartitionState(slice);
}
}
@@ -150,9 +143,9 @@ public class PartitionRecognizer {
return PartitionState.NEED_MORE_DATA_STATE;
} else {
int endPartitionIndex = findNextDifferentRowIndex();
- Iterator<Record> recordIterator = getRecordIterator(currentIndex, endPartitionIndex);
+ Slice slice = getSlice(currentIndex, endPartitionIndex);
currentIndex = endPartitionIndex;
- return PartitionState.newPartitionState(recordIterator);
+ return PartitionState.newPartitionState(slice);
}
}
@@ -179,85 +172,12 @@ public class PartitionRecognizer {
return i;
}
- private Iterator<Record> getRecordIterator(int startPartitionIndex, int endPartitionIndex) {
- return new Iterator<Record>() {
- private int curIndex = startPartitionIndex;
- private final int endIndex = endPartitionIndex;
- private final Column[] columns = currentTsBlock.getColumns(requiredChannels);
-
- @Override
- public boolean hasNext() {
- return curIndex < endIndex;
- }
-
- @Override
- public Record next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- final int idx = curIndex++;
- return new Record() {
- @Override
- public int getInt(int columnIndex) {
- return columns[columnIndex].getInt(idx);
- }
-
- @Override
- public long getLong(int columnIndex) {
- return columns[columnIndex].getLong(idx);
- }
-
- @Override
- public float getFloat(int columnIndex) {
- return columns[columnIndex].getFloat(idx);
- }
-
- @Override
- public double getDouble(int columnIndex) {
- return columns[columnIndex].getDouble(idx);
- }
-
- @Override
- public boolean getBoolean(int columnIndex) {
- return columns[columnIndex].getBoolean(idx);
- }
-
- @Override
- public Binary getBinary(int columnIndex) {
- return columns[columnIndex].getBinary(idx);
- }
-
- @Override
- public String getString(int columnIndex) {
- return columns[columnIndex].getBinary(idx).getStringValue(TSFileConfig.STRING_CHARSET);
- }
-
- @Override
- public LocalDate getLocalDate(int columnIndex) {
- return DateUtils.parseIntToLocalDate(columns[columnIndex].getInt(idx));
- }
-
- @Override
- public Object getObject(int columnIndex) {
- return columns[columnIndex].getObject(idx);
- }
-
- @Override
- public Type getDataType(int columnIndex) {
- return outputDataTypes.get(columnIndex);
- }
-
- @Override
- public boolean isNull(int columnIndex) {
- return columns[columnIndex].isNull(idx);
- }
-
- @Override
- public int size() {
- return columns.length;
- }
- };
- }
- };
+ private Slice getSlice(int startPartitionIndex, int endPartitionIndex) {
+ return new Slice(
+ startPartitionIndex,
+ endPartitionIndex,
+ currentTsBlock.getValueColumns(),
+ requiredChannels,
+ inputDataTypes);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/TableFunctionOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/TableFunctionOperator.java
index 0ad55973e38..58ba5bdac45 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/TableFunctionOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/TableFunctionOperator.java
@@ -23,16 +23,20 @@ import org.apache.iotdb.db.queryengine.execution.operator.Operator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.ProcessOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.PartitionState;
+import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.Slice;
+import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.SliceCache;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionNode;
import org.apache.iotdb.udf.api.relational.access.Record;
import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.LongColumnBuilder;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import java.util.Arrays;
@@ -49,16 +53,22 @@ public class TableFunctionOperator implements ProcessOperator {
private final TableFunctionProcessorProvider processorProvider;
private final PartitionRecognizer partitionRecognizer;
private final TsBlockBuilder blockBuilder;
+ private final int properChannelCount;
+ private final boolean needPassThrough;
private TableFunctionDataProcessor processor;
private PartitionState partitionState;
private ListenableFuture<?> isBlocked;
private boolean finished = false;
+ private ColumnBuilder passThroughIndexBuilder;
+
+ private SliceCache sliceCache;
public TableFunctionOperator(
OperatorContext operatorContext,
TableFunctionProcessorProvider processorProvider,
Operator inputOperator,
+ List<TSDataType> inputDataTypes,
List<TSDataType> outputDataTypes,
int properChannelCount,
List<Integer> requiredChannels,
@@ -66,11 +76,14 @@ public class TableFunctionOperator implements ProcessOperator {
List<Integer> partitionChannels) {
this.operatorContext = operatorContext;
this.inputOperator = inputOperator;
+ this.properChannelCount = properChannelCount;
this.processorProvider = processorProvider;
this.partitionRecognizer =
- new PartitionRecognizer(partitionChannels, requiredChannels, outputDataTypes);
+ new PartitionRecognizer(partitionChannels, requiredChannels, inputDataTypes);
+ this.needPassThrough = properChannelCount != outputDataTypes.size();
this.partitionState = null;
this.blockBuilder = new TsBlockBuilder(outputDataTypes);
+ this.sliceCache = new SliceCache();
}
@Override
@@ -107,42 +120,72 @@ public class TableFunctionOperator implements ProcessOperator {
@Override
public TsBlock next() throws Exception {
PartitionState.StateType stateType = partitionState.getStateType();
- Iterator<Record> recordIterator = partitionState.getRecordIterator();
+ Slice slice = partitionState.getSlice();
partitionState = null;
if (stateType == PartitionState.StateType.INIT
|| stateType == PartitionState.StateType.NEED_MORE_DATA) {
isBlocked = null;
return null;
} else {
- List<ColumnBuilder> columnBuilders = getOutputColumnBuilders();
+ List<ColumnBuilder> properColumnBuilders = getProperColumnBuilders();
+ ColumnBuilder passThroughIndexBuilder = getPassThroughIndexBuilder();
if (stateType == PartitionState.StateType.FINISHED) {
if (processor != null) {
- processor.finish(columnBuilders);
+ processor.finish(properColumnBuilders, passThroughIndexBuilder);
}
finished = true;
- return buildTsBlock(columnBuilders);
+ return buildTsBlock(properColumnBuilders, passThroughIndexBuilder);
}
if (stateType == PartitionState.StateType.NEW_PARTITION) {
if (processor != null) {
- processor.finish(columnBuilders);
+ processor.finish(properColumnBuilders, passThroughIndexBuilder);
}
+ sliceCache.clear();
processor = processorProvider.getDataProcessor();
}
+ sliceCache.addSlice(slice);
+ Iterator<Record> recordIterator = slice.getRequiredRecordIterator();
while (recordIterator.hasNext()) {
- processor.process(recordIterator.next(), columnBuilders);
+ processor.process(recordIterator.next(), properColumnBuilders, passThroughIndexBuilder);
}
- return buildTsBlock(columnBuilders);
+ return buildTsBlock(properColumnBuilders, passThroughIndexBuilder);
}
}
- private List<ColumnBuilder> getOutputColumnBuilders() {
+ private List<ColumnBuilder> getProperColumnBuilders() {
blockBuilder.reset();
- return Arrays.asList(blockBuilder.getValueColumnBuilders());
+ return Arrays.asList(blockBuilder.getValueColumnBuilders()).subList(0, properChannelCount);
+ }
+
+ private ColumnBuilder getPassThroughIndexBuilder() {
+ if (needPassThrough) {
+ passThroughIndexBuilder = new LongColumnBuilder(null, 1);
+ }
+ return passThroughIndexBuilder;
}
- private TsBlock buildTsBlock(List<ColumnBuilder> columnBuilders) {
- int positionCount = columnBuilders.get(0).getPositionCount();
+ private TsBlock buildTsBlock(
+ List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) {
+ List<ColumnBuilder> passThroughColumnBuilders =
+ Arrays.asList(blockBuilder.getValueColumnBuilders())
+ .subList(properChannelCount, blockBuilder.getValueColumnBuilders().length);
+ int positionCount = properColumnBuilders.get(0).getPositionCount();
blockBuilder.declarePositions(positionCount);
+ if (needPassThrough) {
+ // handle pass through column only if needed
+ Column passThroughIndex = passThroughIndexBuilder.build();
+ for (int i = 0; i < passThroughIndex.getPositionCount(); i++) {
+ long index = passThroughIndex.getLong(i);
+ Record row = sliceCache.getOriginalRecord((int) index);
+ for (int j = 0; j < row.size(); j++) {
+ if (row.isNull(j)) {
+ passThroughColumnBuilders.get(j).appendNull();
+ } else {
+ passThroughColumnBuilders.get(j).writeObject(row.getObject(j));
+ }
+ }
+ }
+ }
return blockBuilder.build(new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, positionCount));
}
@@ -150,13 +193,15 @@ public class TableFunctionOperator implements ProcessOperator {
public boolean hasNext() throws Exception {
if (partitionState == null) {
isBlocked().get(); // wait for the next TsBlock
- partitionState = partitionRecognizer.getState();
+ partitionState = partitionRecognizer.nextState();
}
return !finished;
}
@Override
- public void close() throws Exception {}
+ public void close() throws Exception {
+ sliceCache.close();
+ }
@Override
public boolean isFinished() throws Exception {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/PartitionState.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/PartitionState.java
index 979fec9b611..1c3c6e8686a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/PartitionState.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/PartitionState.java
@@ -19,10 +19,6 @@
package org.apache.iotdb.db.queryengine.execution.operator.process.function.partition;
-import org.apache.iotdb.udf.api.relational.access.Record;
-
-import java.util.Iterator;
-
public class PartitionState {
public static final PartitionState NEED_MORE_DATA_STATE =
@@ -30,12 +26,12 @@ public class PartitionState {
public static final PartitionState FINISHED_STATE = new PartitionState(StateType.FINISHED, null);
public static final PartitionState INIT_STATE = new PartitionState(StateType.INIT, null);
- public static PartitionState newPartitionState(Iterator<Record> recordIterator) {
- return new PartitionState(StateType.NEW_PARTITION, recordIterator);
+ public static PartitionState newPartitionState(Slice slice) {
+ return new PartitionState(StateType.NEW_PARTITION, slice);
}
- public static PartitionState iteratingState(Iterator<Record> recordIterator) {
- return new PartitionState(StateType.ITERATING, recordIterator);
+ public static PartitionState iteratingState(Slice slice) {
+ return new PartitionState(StateType.ITERATING, slice);
}
public enum StateType {
@@ -48,11 +44,11 @@ public class PartitionState {
private final StateType stateType;
// Nullable
- private final Iterator<Record> recordIterator;
+ private final Slice slice;
- protected PartitionState(StateType stateType, Iterator<Record> recordIterator) {
+ protected PartitionState(StateType stateType, Slice slice) {
this.stateType = stateType;
- this.recordIterator = recordIterator;
+ this.slice = slice;
}
public StateType getStateType() {
@@ -60,7 +56,7 @@ public class PartitionState {
}
// Nullable
- public Iterator<Record> getRecordIterator() {
- return recordIterator;
+ public Slice getSlice() {
+ return slice;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/Slice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/Slice.java
new file mode 100644
index 00000000000..19b7d47f4e9
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/Slice.java
@@ -0,0 +1,153 @@
+/*
+ * 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.queryengine.execution.operator.process.function.partition;
+
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.DateUtils;
+
+import java.time.LocalDate;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/** Parts of partition. */
+public class Slice {
+
+ private final int startIndex;
+ private final int endIndex;
+ private final Column[] requiredColumns;
+ private final Column[] originalColumns;
+ private final List<Type> dataTypes;
+
+ public Slice(
+ int startIndex,
+ int endIndex,
+ Column[] columns,
+ int[] requiredChannels,
+ List<Type> dataTypes) {
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.originalColumns = columns;
+ this.requiredColumns = new Column[requiredChannels.length];
+ for (int i = 0; i < requiredChannels.length; i++) {
+ requiredColumns[i] = columns[requiredChannels[i]];
+ }
+ this.dataTypes = dataTypes;
+ }
+
+ public int getSize() {
+ return endIndex - startIndex;
+ }
+
+ public Record getOriginalRecord(int offset) {
+ return getRecord(startIndex + offset, originalColumns);
+ }
+
+ public Iterator<Record> getRequiredRecordIterator() {
+ return new Iterator<Record>() {
+ private int curIndex = startIndex;
+
+ @Override
+ public boolean hasNext() {
+ return curIndex < endIndex;
+ }
+
+ @Override
+ public Record next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final int idx = curIndex++;
+ return getRecord(idx, requiredColumns);
+ }
+ };
+ }
+
+ private Record getRecord(int offset, Column[] originalColumns) {
+ return new Record() {
+ @Override
+ public int getInt(int columnIndex) {
+ return originalColumns[columnIndex].getInt(offset);
+ }
+
+ @Override
+ public long getLong(int columnIndex) {
+ return originalColumns[columnIndex].getLong(offset);
+ }
+
+ @Override
+ public float getFloat(int columnIndex) {
+ return originalColumns[columnIndex].getFloat(offset);
+ }
+
+ @Override
+ public double getDouble(int columnIndex) {
+ return originalColumns[columnIndex].getDouble(offset);
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) {
+ return originalColumns[columnIndex].getBoolean(offset);
+ }
+
+ @Override
+ public Binary getBinary(int columnIndex) {
+ return originalColumns[columnIndex].getBinary(offset);
+ }
+
+ @Override
+ public String getString(int columnIndex) {
+ return originalColumns[columnIndex]
+ .getBinary(offset)
+ .getStringValue(TSFileConfig.STRING_CHARSET);
+ }
+
+ @Override
+ public LocalDate getLocalDate(int columnIndex) {
+ return DateUtils.parseIntToLocalDate(originalColumns[columnIndex].getInt(offset));
+ }
+
+ @Override
+ public Object getObject(int columnIndex) {
+ return originalColumns[columnIndex].getObject(offset);
+ }
+
+ @Override
+ public Type getDataType(int columnIndex) {
+ return dataTypes.get(columnIndex);
+ }
+
+ @Override
+ public boolean isNull(int columnIndex) {
+ return originalColumns[columnIndex].isNull(offset);
+ }
+
+ @Override
+ public int size() {
+ return originalColumns.length;
+ }
+ };
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/SliceCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/SliceCache.java
new file mode 100644
index 00000000000..993289fd1a9
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/function/partition/SliceCache.java
@@ -0,0 +1,55 @@
+/*
+ * 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.queryengine.execution.operator.process.function.partition;
+
+import org.apache.iotdb.udf.api.relational.access.Record;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Used to manage the slices of the partition. It is all in memory now. */
+public class SliceCache {
+
+ private final List<Slice> slices = new ArrayList<>();
+
+ public Record getOriginalRecord(long index) {
+ long previousSize = 0;
+ for (Slice slice : slices) {
+ long currentSize = slice.getSize();
+ if (index < previousSize + currentSize) {
+ return slice.getOriginalRecord((int) (index - previousSize));
+ }
+ previousSize += currentSize;
+ }
+ throw new IndexOutOfBoundsException("Index out of bound");
+ }
+
+ public void addSlice(Slice slice) {
+ slices.add(slice);
+ }
+
+ public void clear() {
+ slices.clear();
+ }
+
+ public void close() {
+ // do nothing
+ }
+}
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 f28c338b6ac..6abe443616c 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
@@ -2353,6 +2353,12 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
node.getPlanNodeId(),
TableFunctionOperator.class.getSimpleName());
+ List<TSDataType> inputDataTypes =
+ node.getChild().getOutputSymbols().stream()
+ .map(context.getTypeProvider()::getTableModelType)
+ .map(InternalTypeManager::getTSDataType)
+ .collect(Collectors.toList());
+
List<TSDataType> outputDataTypes =
node.getOutputSymbols().stream()
.map(context.getTypeProvider()::getTableModelType)
@@ -2380,6 +2386,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
operatorContext,
processorProvider,
operator,
+ inputDataTypes,
outputDataTypes,
properChannelCount,
requiredChannels,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index 441b31b2c42..3fab6cfd6ea 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
@@ -78,7 +78,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNod
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.udf.api.relational.table.argument.Argument;
-import org.apache.iotdb.udf.api.relational.table.argument.DescriptorArgument;
import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
@@ -96,7 +95,6 @@ import java.util.Optional;
import static com.google.common.base.Preconditions.checkArgument;
import static java.lang.String.format;
-import static java.util.stream.Collectors.joining;
import static org.apache.iotdb.db.utils.DateTimeUtils.TIMESTAMP_PRECISION;
public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter.GraphContext> {
@@ -989,11 +987,10 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
private String formatArgument(String argumentName, Argument argument) {
if (argument instanceof ScalarArgument) {
return formatScalarArgument(argumentName, (ScalarArgument) argument);
- }
- if (argument instanceof DescriptorArgument) {
- return formatDescriptorArgument(argumentName, (DescriptorArgument) argument);
- } else {
+ } else if (argument instanceof TableArgument) {
return formatTableArgument(argumentName, (TableArgument) argument);
+ } else {
+ return argumentName + " => " + argument;
}
}
@@ -1003,19 +1000,6 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
argumentName, argument.getType(), argument.getValue());
}
- private String formatDescriptorArgument(String argumentName, DescriptorArgument argument) {
- String descriptor;
- if (argument.getDescriptor().isPresent()) {
- descriptor =
- argument.getDescriptor().get().getFields().stream()
- .map(field -> field.getName() + field.getType().map(type -> " " + type).orElse(""))
- .collect(joining(", ", "(", ")"));
- } else {
- descriptor = "NULL";
- }
- return format("%s => DescriptorArgument{%s}", argumentName, descriptor);
- }
-
private String formatTableArgument(String argumentName, TableArgument argument) {
return format("%s => TableArgument", argumentName);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index d79fcca4476..81ec55af5b0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@ -140,7 +140,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SubqueryExpressio
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SymbolReference;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Table;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionArgument;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionInvocation;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionTableArgument;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableSubquery;
@@ -154,7 +153,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.WithQuery;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.WrappedInsertStatement;
import org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
import org.apache.iotdb.db.queryengine.plan.relational.type.TypeManager;
-import org.apache.iotdb.db.queryengine.plan.relational.type.TypeNotFoundException;
import org.apache.iotdb.db.queryengine.plan.statement.component.FillPolicy;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
@@ -167,11 +165,8 @@ import org.apache.iotdb.udf.api.relational.TableFunction;
import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
import org.apache.iotdb.udf.api.relational.table.argument.Argument;
import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
-import org.apache.iotdb.udf.api.relational.table.argument.Descriptor;
-import org.apache.iotdb.udf.api.relational.table.argument.DescriptorArgument;
import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
-import org.apache.iotdb.udf.api.relational.table.specification.DescriptorParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
@@ -238,10 +233,8 @@ import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Join.Type.
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Join.Type.LEFT;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Join.Type.RIGHT;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.util.AstUtil.preOrder;
-import static org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignatureTranslator.toTypeSignature;
import static org.apache.iotdb.db.queryengine.plan.relational.utils.NodeUtils.getSortItemsFromOrderBy;
import static org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS;
-import static org.apache.iotdb.udf.api.relational.table.argument.DescriptorArgument.NULL_DESCRIPTOR;
import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN;
public class StatementAnalyzer {
@@ -3139,12 +3132,16 @@ public class StatementAnalyzer {
ArgumentsAnalysis argumentsAnalysis =
analyzeArguments(
- function.getArgumentsSpecification(), node.getArguments(), scope, errorLocation);
+ function.getArgumentsSpecifications(), node.getArguments(), scope, errorLocation);
TableFunctionAnalysis functionAnalysis =
function.analyze(argumentsAnalysis.getPassedArguments());
- if (node.getCopartitioning().size() != 0) {
- // TODO(UDF): support copartition in future
+ // At most one table argument can be passed to a table function now
+ if (argumentsAnalysis.getTableArgumentAnalyses().size() > 1) {
+ throw new SemanticException("At most one table argument can be passed to a table function");
+ }
+ // Copartition is not supported now
+ if (!node.getCopartitioning().isEmpty()) {
throw new SemanticException("Copartitioning is not supported now.");
}
@@ -3163,15 +3160,19 @@ public class StatementAnalyzer {
"Table function %s specifies required columns from table argument %s which cannot be found",
node.getName(), name));
}
- // TODO(UDF): 定义的时候需要确保 columns 不是empty且不为负数，不在这里二次确认了
- // if (columns.isEmpty()) {
- // throw new TrinoException(
- // FUNCTION_IMPLEMENTATION_ERROR,
- // format(
- // "Table function %s specifies empty list of required columns from
- // table argument %s",
- // node.getName(), name));
- // }
+ // make sure the required columns are not empty and positive
+ if (columns.isEmpty()) {
+ throw new SemanticException(
+ String.format(
+ "Table function %s specifies empty list of required columns from table argument %s",
+ node.getName(), name));
+ }
+ if (columns.stream().anyMatch(column -> column < 0)) {
+ throw new SemanticException(
+ String.format(
+ "Table function %s specifies negative index of required column from table argument %s",
+ node.getName(), name));
+ }
// the scope is recorded, because table arguments are already analyzed
Scope inputScope = analysis.getScope(tableArgumentsByName.get(name).getRelation());
columns.stream()
@@ -3189,8 +3190,7 @@ public class StatementAnalyzer {
.map(inputScope.getRelationType()::getFieldByIndex)
.forEach(this::recordColumnAccess);
});
- // TODO(UDF)：我觉得requiredInputs应该是一定等于tableArgumentNameSet的。可以考虑让用户在定义的时候，一定要为每个table
- // argument指定required columns？这个应该在注册的时候做检查。
+ // check that all required inputs are specified
Set<String> requiredInputs = ImmutableSet.copyOf(requiredColumns.keySet());
tableArgumentNameSet.stream()
.filter(input -> !requiredInputs.contains(input))
@@ -3224,7 +3224,7 @@ public class StatementAnalyzer {
// next, columns derived from table arguments, in order of argument declarations
List<String> tableArgumentNames =
- function.getArgumentsSpecification().stream()
+ function.getArgumentsSpecifications().stream()
.filter(TableParameterSpecification.class::isInstance)
.map(ParameterSpecification::getName)
.collect(toImmutableList());
@@ -3288,7 +3288,6 @@ public class StatementAnalyzer {
ImmutableMap.Builder<String, Argument> passedArguments = ImmutableMap.builder();
ImmutableList.Builder<TableArgumentAnalysis> tableArgumentAnalyses = ImmutableList.builder();
// TODO(UDF): args passed positionally - can one only pass some prefix of args?
- // TODO(UDF): check at most one table argument passed
if (argumentsPassedByName) {
Map<String, ParameterSpecification> argumentSpecificationsByName = new HashMap<>();
for (ParameterSpecification parameterSpecification : parameterSpecifications) {
@@ -3344,7 +3343,6 @@ public class StatementAnalyzer {
analyzeDefault(parameterSpecification, errorLocation));
}
}
-
return new ArgumentsAnalysis(passedArguments.buildOrThrow(), tableArgumentAnalyses.build());
}
@@ -3355,8 +3353,6 @@ public class StatementAnalyzer {
String actualType;
if (argument.getValue() instanceof TableFunctionTableArgument) {
actualType = "table";
- } else if (argument.getValue() instanceof TableFunctionDescriptorArgument) {
- actualType = "descriptor";
} else if (argument.getValue() instanceof Expression) {
actualType = "expression";
} else {
@@ -3377,14 +3373,6 @@ public class StatementAnalyzer {
(TableFunctionTableArgument) argument.getValue(),
(TableParameterSpecification) parameterSpecification,
scope);
- } else if (parameterSpecification instanceof DescriptorParameterSpecification) {
- if (!(argument.getValue() instanceof TableFunctionDescriptorArgument)) {
- throw new SemanticException(
- String.format(
- "Invalid argument %s. Expected descriptor argument, got %s",
- parameterSpecification.getName(), actualType));
- }
- return analyzeDescriptorArgument((TableFunctionDescriptorArgument) argument.getValue());
} else if (parameterSpecification instanceof ScalarParameterSpecification) {
if (!(argument.getValue() instanceof Expression)) {
throw new SemanticException(
@@ -3392,15 +3380,6 @@ public class StatementAnalyzer {
"Invalid argument %s. Expected scalar argument, got %s",
parameterSpecification.getName(), actualType));
}
- // TODO(UDF):
- // 如果functionName是descriptor的话，会被解析成FunctionCall还是TableFunctionDescriptorArgument？
- // // 'descriptor' as a function name is not allowed in this context
- // if (expression instanceof FunctionCall && ((FunctionCall)
- // expression).getName().hasSuffix(QualifiedName.of("descriptor"))) { // function name is
- // always compared case-insensitive
- // throw semanticException(INVALID_FUNCTION_ARGUMENT, argument, "'descriptor'
- // function is not allowed as a table function argument");
- // }
return analyzeScalarArgument(
(Expression) argument.getValue(),
(ScalarParameterSpecification) parameterSpecification);
@@ -3543,40 +3522,6 @@ public class StatementAnalyzer {
Optional.of(analysisBuilder.build()));
}
- private ArgumentAnalysis analyzeDescriptorArgument(TableFunctionDescriptorArgument argument) {
- return new ArgumentAnalysis(
- argument
- .getDescriptor()
- .map(
- descriptor ->
- new DescriptorArgument(
- Optional.of(
- new Descriptor(
- descriptor.getFields().stream()
- .map(
- field ->
- new Descriptor.Field(
- field.getName().getCanonicalValue(),
- field
- .getType()
- .map(
- type -> {
- try {
- return UDFDataTypeTransformer
- .transformReadTypeToUDFDataType(
- typeManager.getType(
- toTypeSignature(type)));
- } catch (TypeNotFoundException e) {
- throw new SemanticException(
- String.format(
- "Unknown type: %s", type));
- }
- })))
- .collect(toImmutableList())))))
- .orElse(NULL_DESCRIPTOR),
- Optional.empty());
- }
-
private ArgumentAnalysis analyzeScalarArgument(
Expression expression, ScalarParameterSpecification argumentSpecification) {
// currently, only constant arguments are supported
@@ -3609,14 +3554,7 @@ public class StatementAnalyzer {
!(parameterSpecification instanceof TableParameterSpecification),
"Table argument specification cannot have a default value.");
- if (parameterSpecification instanceof DescriptorParameterSpecification) {
- if (parameterSpecification.getDefaultValue().isPresent()) {
- return new DescriptorArgument(
- Optional.of((Descriptor) parameterSpecification.getDefaultValue().get()));
- } else {
- return NULL_DESCRIPTOR;
- }
- } else if (parameterSpecification instanceof ScalarParameterSpecification) {
+ if (parameterSpecification instanceof ScalarParameterSpecification) {
checkArgument(
parameterSpecification.getDefaultValue().isPresent(),
String.format(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
index 344dad521a7..42ca79d3a1c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
@@ -633,14 +633,6 @@ public abstract class AstVisitor<R, C> {
return visitNode(tableFunctionTableArgument, context);
}
- public R visitDescriptorField(DescriptorField descriptorField, C context) {
- return visitNode(descriptorField, context);
- }
-
- public R visitDescriptor(Descriptor descriptor, C context) {
- return visitNode(descriptor, context);
- }
-
public R visitEmptyTableTreatment(EmptyTableTreatment emptyTableTreatment, C context) {
return visitNode(emptyTableTreatment, context);
}
@@ -649,11 +641,6 @@ public abstract class AstVisitor<R, C> {
return visitNode(tableFunctionArgument, context);
}
- public R visitDescriptorArgument(
- TableFunctionDescriptorArgument tableFunctionDescriptorArgument, C context) {
- return visitNode(tableFunctionDescriptorArgument, context);
- }
-
public R visitTableFunctionInvocation(
TableFunctionInvocation tableFunctionInvocation, C context) {
return visitNode(tableFunctionInvocation, context);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/Descriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/Descriptor.java
deleted file mode 100644
index d86c86b1b50..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/Descriptor.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.queryengine.plan.relational.sql.ast;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static java.util.Objects.requireNonNull;
-
-public class Descriptor extends Node {
- private final List<DescriptorField> fields;
-
- public Descriptor(NodeLocation location, List<DescriptorField> fields) {
- super(location);
- requireNonNull(fields, "fields is null");
- checkArgument(!fields.isEmpty(), "fields list is empty");
- this.fields = fields;
- }
-
- public List<DescriptorField> getFields() {
- return fields;
- }
-
- @Override
- public <R, C> R accept(AstVisitor<R, C> visitor, C context) {
- return visitor.visitDescriptor(this, context);
- }
-
- @Override
- public List<? extends Node> getChildren() {
- return fields;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- return Objects.equals(fields, ((Descriptor) o).fields);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(fields);
- }
-
- @Override
- public String toString() {
- return fields.stream()
- .map(DescriptorField::toString)
- .collect(Collectors.joining(", ", "DESCRIPTOR(", ")"));
- }
-
- @Override
- public boolean shallowEquals(Node o) {
- return sameClass(this, o);
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DescriptorField.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DescriptorField.java
deleted file mode 100644
index 09c8f6acdb2..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DescriptorField.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.queryengine.plan.relational.sql.ast;
-
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-
-import static java.util.Objects.requireNonNull;
-
-public class DescriptorField extends Node {
- private final Identifier name;
- private final Optional<DataType> type;
-
- public DescriptorField(NodeLocation location, Identifier name, Optional<DataType> type) {
- super(location);
- this.name = requireNonNull(name, "name is null");
- this.type = requireNonNull(type, "type is null");
- }
-
- public Identifier getName() {
- return name;
- }
-
- public Optional<DataType> getType() {
- return type;
- }
-
- @Override
- public <R, C> R accept(AstVisitor<R, C> visitor, C context) {
- return visitor.visitDescriptorField(this, context);
- }
-
- @Override
- public List<? extends Node> getChildren() {
- return type.map(ImmutableList::of).orElse(ImmutableList.of());
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- DescriptorField field = (DescriptorField) o;
- return Objects.equals(name, field.name) && Objects.equals(type, field.type);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(name, type);
- }
-
- @Override
- public String toString() {
- return type.map(dataType -> name + " " + dataType).orElseGet(name::toString);
- }
-
- @Override
- public boolean shallowEquals(Node o) {
- if (!sameClass(this, o)) {
- return false;
- }
-
- return Objects.equals(name, ((DescriptorField) o).name);
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionArgument.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionArgument.java
index 01642378f60..5fac31abc43 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionArgument.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionArgument.java
@@ -36,10 +36,7 @@ public class TableFunctionArgument extends Node {
super(location);
this.name = requireNonNull(name, "name is null");
requireNonNull(value, "value is null");
- checkArgument(
- value instanceof TableFunctionTableArgument
- || value instanceof TableFunctionDescriptorArgument
- || value instanceof Expression);
+ checkArgument(value instanceof TableFunctionTableArgument || value instanceof Expression);
this.value = value;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionDescriptorArgument.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionDescriptorArgument.java
deleted file mode 100644
index 65ac6d32391..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/TableFunctionDescriptorArgument.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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.queryengine.plan.relational.sql.ast;
-
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-
-import static java.util.Objects.requireNonNull;
-
-public class TableFunctionDescriptorArgument extends Node {
- private final Optional<Descriptor> descriptor;
-
- public static TableFunctionDescriptorArgument descriptorArgument(
- NodeLocation location, Descriptor descriptor) {
- requireNonNull(descriptor, "descriptor is null");
- return new TableFunctionDescriptorArgument(location, Optional.of(descriptor));
- }
-
- public static TableFunctionDescriptorArgument nullDescriptorArgument(NodeLocation location) {
- return new TableFunctionDescriptorArgument(location, Optional.empty());
- }
-
- private TableFunctionDescriptorArgument(NodeLocation location, Optional<Descriptor> descriptor) {
- super(location);
- this.descriptor = descriptor;
- }
-
- public Optional<Descriptor> getDescriptor() {
- return descriptor;
- }
-
- @Override
- public <R, C> R accept(AstVisitor<R, C> visitor, C context) {
- return visitor.visitDescriptorArgument(this, context);
- }
-
- @Override
- public List<? extends Node> getChildren() {
- return descriptor.map(ImmutableList::of).orElse(ImmutableList.of());
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- return Objects.equals(descriptor, ((TableFunctionDescriptorArgument) o).descriptor);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(descriptor);
- }
-
- @Override
- public String toString() {
- return descriptor.map(Descriptor::toString).orElse("CAST (NULL AS DESCRIPTOR)");
- }
-
- @Override
- public boolean shallowEquals(Node o) {
- return sameClass(this, o);
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index 26251e8a117..83351aa3562 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -67,8 +67,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DeleteDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DereferenceExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Descriptor;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescriptorField;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DoubleLiteral;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
@@ -246,8 +244,6 @@ import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.GroupingSe
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.GroupingSets.Type.EXPLICIT;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.GroupingSets.Type.ROLLUP;
import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.QualifiedName.mapIdentifier;
-import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument.descriptorArgument;
-import static org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument.nullDescriptorArgument;
import static org.apache.iotdb.db.utils.TimestampPrecisionUtils.currPrecision;
import static org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_AGGREGATION;
import static org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_BY_AGGREGATION;
@@ -1864,8 +1860,6 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
Node value;
if (context.tableArgument() != null) {
value = visit(context.tableArgument());
- } else if (context.descriptorArgument() != null) {
- value = visit(context.descriptorArgument());
} else {
value = visit(context.scalarArgument());
}
@@ -1946,24 +1940,6 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
return relation;
}
- @Override
- public Node visitDescriptorArgument(RelationalSqlParser.DescriptorArgumentContext context) {
- if (context.NULL() != null) {
- return nullDescriptorArgument(getLocation(context));
- }
- List<DescriptorField> fields = visit(context.descriptorField(), DescriptorField.class);
- return descriptorArgument(
- getLocation(context), new Descriptor(getLocation(context.DESCRIPTOR()), fields));
- }
-
- @Override
- public Node visitDescriptorField(RelationalSqlParser.DescriptorFieldContext context) {
- return new DescriptorField(
- getLocation(context),
- (Identifier) visit(context.identifier()),
- visitIfPresent(context.type(), DataType.class));
- }
-
@Override
public Node visitScalarArgument(RelationalSqlParser.ScalarArgumentContext ctx) {
if (ctx.expression() != null) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/SqlFormatter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/SqlFormatter.java
index 5f85e585db1..eed5429e773 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/SqlFormatter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/SqlFormatter.java
@@ -88,7 +88,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.StartPipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.StopPipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Table;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionArgument;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionDescriptorArgument;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionInvocation;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableFunctionTableArgument;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TableSubquery;
@@ -1315,27 +1314,5 @@ public final class SqlFormatter {
return null;
}
-
- @Override
- public Void visitDescriptorArgument(TableFunctionDescriptorArgument node, Integer indent) {
- if (node.getDescriptor().isPresent()) {
- builder.append(
- node.getDescriptor().get().getFields().stream()
- .map(
- field -> {
- String formattedField = formatName(field.getName());
- if (field.getType().isPresent()) {
- formattedField =
- formattedField + " " + formatExpression(field.getType().get());
- }
- return formattedField;
- })
- .collect(joining(", ", "DESCRIPTOR(", ")")));
- } else {
- builder.append("CAST (NULL AS DESCRIPTOR)");
- }
-
- return null;
- }
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java
index 70fe70eaf85..90dfcb673fb 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.queryengine.execution.operator.Operator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.function.PartitionRecognizer;
import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.PartitionState;
+import org.apache.iotdb.db.queryengine.execution.operator.process.function.partition.Slice;
import org.apache.iotdb.udf.api.relational.access.Record;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -187,7 +188,7 @@ public class TableFunctionOperatorTest {
}
@Test
- public void PartitionRecognizerTest() {
+ public void testPartitionRecognizer() {
QueryId queryId = new QueryId("stub_query");
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance");
@@ -202,91 +203,97 @@ public class TableFunctionOperatorTest {
Collections.singletonList(1),
Arrays.asList(0, 1, 3),
Arrays.asList(TSDataType.TIMESTAMP, TSDataType.STRING, TSDataType.INT32));
- PartitionState state = partitionRecognizer.getState();
+ PartitionState state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.INIT_STATE, state);
- Assert.assertEquals(PartitionState.INIT_STATE, partitionRecognizer.getState());
+ Assert.assertEquals(PartitionState.INIT_STATE, partitionRecognizer.nextState());
// 1. add first TsBlock, expected NEW_PARTITION and NEED_MORE_DATA
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEW_PARTITION, state.getStateType());
- checkIteratorSimply(state.getRecordIterator(), Collections.emptyList());
- state = partitionRecognizer.getState();
+ checkIteratorSimply(state.getSlice(), Collections.emptyList());
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEW_PARTITION, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Arrays.asList(
Arrays.asList(1717171200000L, D1_BINARY, 60),
Arrays.asList(1719763200000L, D1_BINARY, 60)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
// 2. add second TsBlock, expected ITERATING, NEW_PARTITION and NEED_MORE_DATA
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.ITERATING, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Collections.singletonList(Arrays.asList(1722441600000L, D1_BINARY, 61)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEW_PARTITION, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Arrays.asList(
Arrays.asList(1725120000000L, D2_BINARY, 62),
Arrays.asList(1725130000000L, D2_BINARY, 62),
Arrays.asList(1725140000000L, D2_BINARY, 63)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
// 3. add third and fourth TsBlock, expected NEED_MORE_DATA
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
// 4. add fifth TsBlock, expected NEW_PARTITION, NEW_PARTITION and NEED_MORE_DATA
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEW_PARTITION, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Arrays.asList(
Arrays.asList(1722441600000L, D3_BINARY, 63),
Arrays.asList(1722441800000L, D3_BINARY, 64)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEW_PARTITION, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Collections.singletonList(Arrays.asList(1722551600000L, D5_BINARY, 64)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
// 5. add sixth TsBlock, expected ITERATING and NEED_MORE_DATA
Assert.assertTrue(childOperator.hasNext());
partitionRecognizer.addTsBlock(childOperator.next());
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.ITERATING, state.getStateType());
checkIteratorSimply(
- state.getRecordIterator(),
+ state.getSlice(),
Arrays.asList(
Arrays.asList(1722552800000L, D5_BINARY, 65),
Arrays.asList(1722553000000L, D5_BINARY, 65)));
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.NEED_MORE_DATA, state.getStateType());
// no more data, expected FINISHED
Assert.assertFalse(childOperator.hasNext());
partitionRecognizer.noMoreData();
- state = partitionRecognizer.getState();
+ state = partitionRecognizer.nextState();
Assert.assertEquals(PartitionState.StateType.FINISHED, state.getStateType());
} catch (Exception e) {
fail(e.getMessage());
}
}
- private void checkIteratorSimply(Iterator<Record> recordIterable, List<List<Object>> expected) {
+ @Test
+ public void testConstructResult() {
+ // TODO(UDF)
+ }
+
+ private void checkIteratorSimply(Slice slice, List<List<Object>> expected) {
+ Iterator<Record> recordIterable = slice.getRequiredRecordIterator();
int i = 0;
while (recordIterable.hasNext()) {
Record record = recordIterable.next();
diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
index 11bb0ffe28d..656894aba19 100644
--- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
+++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
@@ -740,7 +740,7 @@ tableFunctionCall
;
tableFunctionArgument
- : (identifier '=>')? (tableArgument | descriptorArgument | scalarArgument) // descriptor before expression to avoid parsing descriptor as a function call
+ : (identifier '=>')? (tableArgument | scalarArgument) // descriptor before expression to avoid parsing descriptor as a function call
;
tableArgument
@@ -755,15 +755,6 @@ tableArgumentRelation
| TABLE '(' query ')' (AS? identifier columnAliases?)? #tableArgumentQuery
;
-descriptorArgument
- : DESCRIPTOR '(' descriptorField (',' descriptorField)* ')'
- | CAST '(' NULL AS DESCRIPTOR ')'
- ;
-
-descriptorField
- : identifier type?
- ;
-
scalarArgument
: expression
| timeDuration
(iotdb) 02/03: merge master
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udtf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 602a726ae635d3d03c685cc5d999f2b11db518a7
Merge: 87fc892d072 263443dcb53
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Mon Feb 3 16:06:46 2025 +0800
merge master
.github/workflows/cluster-it-1c1d.yml | 2 +-
.github/workflows/cluster-it-1c1d1a.yml | 2 +-
.github/workflows/cluster-it-1c3d.yml | 2 +-
.github/workflows/compile-check.yml | 2 +-
.github/workflows/daily-it.yml | 2 +-
.github/workflows/daily-ut.yml | 2 +-
.github/workflows/dependency-check.yml | 2 +-
.github/workflows/pipe-it-2cluster.yml | 2 +-
.github/workflows/sonar-codecov.yml | 2 +-
.github/workflows/table-cluster-it-1c1d.yml | 2 +-
.github/workflows/table-cluster-it-1c3d.yml | 2 +-
.github/workflows/unit-test.yml | 2 +-
.github/workflows/vulnerability-check.yml | 2 +-
.mvn/develocity.xml | 3 +-
.mvn/extensions.xml | 4 +-
.../java/org/apache/iotdb/CountPointProcessor.java | 2 +
.../iotdb/AlignedTimeseriesSessionExample.java | 67 +-
.../org/apache/iotdb/SessionConcurrentExample.java | 23 -
.../main/java/org/apache/iotdb/SessionExample.java | 107 +---
.../org/apache/iotdb/trigger/LoggerTrigger.java | 12 +-
.../iotdb/trigger/StatisticsUpdaterTrigger.java | 4 +-
.../apache/iotdb/udf/AggregateFunctionExample.java | 35 +-
.../apache/iotdb/udf/ScalarFunctionExample.java | 57 +-
.../db/query/udf/example/relational/AllSum.java | 49 +-
.../query/udf/example/relational/ContainNull.java | 21 +-
.../relational/{DatePlusOne.java => DatePlus.java} | 31 +-
.../query/udf/example/relational/FirstTwoSum.java | 33 +-
.../db/query/udf/example/relational/MyAvg.java | 32 +-
.../db/query/udf/example/relational/MyCount.java | 19 +-
.../java/org/apache/iotdb/it/env/EnvFactory.java | 5 +
.../iotdb/it/env/cluster/env/AbstractEnv.java | 49 ++
.../iotdb/it/env/cluster/node/AINodeWrapper.java | 5 +
.../it/env/cluster/node/AbstractNodeWrapper.java | 12 +
.../it/env/cluster/node/ConfigNodeWrapper.java | 5 +
.../iotdb/it/env/cluster/node/DataNodeWrapper.java | 5 +
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 12 +
.../org/apache/iotdb/it/utils/TsFileGenerator.java | 87 +--
.../iotdb/it/utils/TsFileTableGenerator.java | 186 ++++++
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 7 +
...ionMigrateDataNodeCrashITFrameworkForIoTV1.java | 2 +-
...ionMigrateDataNodeCrashITFrameworkForIoTV2.java | 2 +-
...oTDBRegionOperationReliabilityITFramework.java} | 174 ++++--
.../IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java | 157 +++++
.../commit/IoTDBRegionReconstructForIoTV1IT.java | 141 +++++
.../IoTDBRegionMigrateNormalITForIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateOtherITForIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateNormalITForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateOtherITForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateDataNodeCrashForIoTV1IT.java | 4 +-
...TDBRegionMigrateDataNodeCrashForIoTV2Batch.java | 4 +-
...DBRegionMigrateDataNodeCrashForIoTV2Stream.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV1IT.java | 4 +-
.../IoTDBRegionMigrateConfigNodeCrashIoTV1IT.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV2Batch.java | 4 +-
...oTDBRegionMigrateConfigNodeCrashIoTV2Batch.java | 4 +-
.../IoTDBRegionMigrateClusterCrashIoTV2Stream.java | 4 +-
...TDBRegionMigrateConfigNodeCrashIoTV2Stream.java | 4 +-
.../IoTDBRemoveDataNodeITFramework.java | 22 +-
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 80 ++-
.../iotdb/db/it/IoTDBLoadTsFileWithModIT.java | 172 ++++++
.../it/IoTDBSyntaxConventionStringLiteralIT.java | 3 +-
.../org/apache/iotdb/db/it/utils/TestUtils.java | 279 ++++++++-
.../iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java | 3 +-
.../pipe/it/autocreate/IoTDBPipeAutoDropIT.java | 10 +-
.../pipe/it/autocreate/IoTDBPipeClusterIT.java | 98 +++
.../pipe/it/autocreate/IoTDBPipeIdempotentIT.java | 93 +++
.../pipe/it/autocreate/IoTDBPipeNullValueIT.java | 28 +-
.../it/autocreate/IoTDBPipeSwitchStatusIT.java | 23 +-
.../pipe/it/manual/IoTDBPipeReqAutoSliceIT.java | 8 +-
.../pipe/it/manual/IoTDBPipeTableManualIT.java | 293 +++++++++
.../manual/IoTDBPipeTypeConversionISessionIT.java | 10 +-
.../iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java | 30 +-
.../pipe/it/tablemodel/IoTDBPipeAutoDropIT.java | 10 +-
.../pipe/it/tablemodel/IoTDBPipeClusterIT.java | 65 ++
.../pipe/it/tablemodel/IoTDBPipeDataSinkIT.java | 450 +++++++++++++-
.../it/tablemodel/IoTDBPipeDoubleLivingIT.java | 329 ++++++++++
.../pipe/it/tablemodel/IoTDBPipeIsolationIT.java | 377 ++++++++++++
.../pipe/it/tablemodel/IoTDBPipeNullValueIT.java | 4 +-
.../pipe/it/tablemodel/IoTDBPipeProtocolIT.java | 1 -
.../it/tablemodel/IoTDBPipeSwitchStatusIT.java | 23 +-
.../pipe/it/tablemodel/IoTDBPipeSyntaxIT.java | 35 +-
.../IoTDBPipeTypeConversionISessionIT.java | 97 ++-
.../pipe/it/tablemodel/IoTDBPipeWithLoadIT.java | 8 +-
.../iotdb/pipe/it/tablemodel/TableModelUtils.java | 614 +++++++++++++++----
.../relational/it/db/it/IoTDBDeletionTableIT.java | 2 +-
.../relational/it/db/it/IoTDBInsertTableIT.java | 2 +-
.../it/db/it/udf/IoTDBSQLFunctionManagementIT.java | 19 +-
.../udf/IoTDBUserDefinedAggregateFunctionIT.java | 53 +-
.../it/udf/IoTDBUserDefinedScalarFunctionIT.java | 2 +-
.../IoTDBAlignByDeviceWithTemplateTableIT.java | 4 +-
.../scalar/IoTDBFormatFunctionTableIT.java | 21 +-
...oTDBNoSelectExpressionAfterAnalyzedTableIT.java | 6 +-
.../query/old/query/IoTDBNullValueFillTableIT.java | 327 ----------
.../it/query/old/query/IoTDBResultSetTableIT.java | 2 +-
.../query/old/query/IoTDBSelectSchemaTableIT.java | 22 +-
.../it/query/recent/IoTDBTableAggregationIT.java | 656 +++++++++++++++++++-
.../recent/subquery/SubqueryDataSetUtils.java | 12 +-
.../IoTDBUncorrelatedInPredicateSubqueryIT.java | 319 ++++++++++
.../IoTDBUncorrelatedQuantifiedComparisonIT.java | 674 +++++++++++++++++++++
.../IoTDBUncorrelatedScalarSubqueryIT.java | 4 +-
.../it/rest/it/IoTDBRestServiceCaseWhenThenIT.java | 7 +-
.../it/rest/it/IoTDBRestServiceFlushQueryIT.java | 6 +-
.../relational/it/rest/it/IoTDBRestServiceIT.java | 3 +-
...IT.java => IoTDBRestServiceInsertValuesIT.java} | 20 +-
.../relational/it/schema/IoTDBDatabaseIT.java | 41 ++
.../iotdb/relational/it/schema/IoTDBDeviceIT.java | 47 +-
.../iotdb/relational/it/schema/IoTDBTableIT.java | 122 ++--
.../it/session/IoTDBSessionRelationalIT.java | 29 +-
.../iotdb/session/it/IoTDBSessionComplexIT.java | 8 +-
.../iotdb/session/it/IoTDBSessionSimpleIT.java | 78 ++-
.../session/it/IoTDBSessionSyntaxConventionIT.java | 18 +-
.../org/apache/iotdb/session/it/SessionIT.java | 32 +-
.../it/local/IoTDBSubscriptionDataTypeIT.java | 8 +-
.../java/org/apache/iotdb/util/MagicUtils.java | 61 ++
.../java/org/apache/iotdb/pipe/api/PipePlugin.java | 37 ++
.../iotdb/pipe/api/annotation/TableModel.java | 23 +-
.../iotdb/pipe/api/annotation/TreeModel.java | 23 +-
.../parameter/PipeParameterValidator.java | 42 +-
.../analysis/AggregateFunctionAnalysis.java | 62 ++
.../api/customizer/analysis/FunctionAnalysis.java} | 8 +-
.../analysis/ScalarFunctionAnalysis.java | 51 ++
...ctionParameters.java => FunctionArguments.java} | 28 +-
.../exception/UDFArgumentNotValidException.java} | 9 +-
.../udf/api/relational/AggregateFunction.java | 48 +-
.../iotdb/udf/api/relational/ScalarFunction.java | 47 +-
.../apache/iotdb/tool/data/AbstractDataTool.java | 32 +-
.../org/apache/iotdb/tool/data/ExportData.java | 43 +-
.../org/apache/iotdb/tool/data/ImportData.java | 24 +-
.../org/apache/iotdb/tool/tsfile/ExportTsFile.java | 3 -
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 3 +
.../java/org/apache/iotdb/session/Session.java | 27 +-
.../payload/SubscriptionFileHandler.java | 24 +-
.../payload/SubscriptionSessionDataSet.java | 12 +-
.../org/apache/iotdb/session/util/RetryUtils.java | 30 +-
.../apache/iotdb/session/util/SessionUtils.java | 25 +-
.../java/org/apache/iotdb/session/TabletTest.java | 28 +-
.../iotdb/session/util/SessionUtilsTest.java | 16 +-
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 16 +-
.../antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4 | 8 +
.../consensus/request/ConfigPhysicalPlan.java | 10 +-
.../consensus/request/ConfigPhysicalPlanType.java | 4 +
.../request/ConfigPhysicalPlanVisitor.java | 110 +++-
.../table/DescTable4InformationSchemaPlan.java} | 12 +-
.../table/ShowTable4InformationSchemaPlan.java} | 16 +-
.../write/pipe/payload/PipeCreateTablePlan.java} | 18 +-
.../pipe/payload/PipeDeactivateTemplatePlan.java | 22 +-
.../write/pipe/payload/PipeDeleteDevicesPlan.java | 130 ++++
.../request/write/table/PreCreateTablePlan.java | 11 +-
.../response/pipe/plugin/PipePluginTableResp.java | 27 +-
.../response/pipe/task/PipeTableResp.java | 7 +
.../DescTable4InformationSchemaResp.java} | 29 +-
.../ShowTable4InformationSchemaResp.java} | 29 +-
.../iotdb/confignode/manager/ClusterManager.java | 2 +-
.../iotdb/confignode/manager/ConfigManager.java | 66 +-
.../apache/iotdb/confignode/manager/IManager.java | 42 +-
.../iotdb/confignode/manager/ProcedureManager.java | 491 +++++++++++----
.../payload/PipeTransferConfigSnapshotSealReq.java | 32 +-
.../protocol/IoTDBConfigRegionAirGapConnector.java | 8 +
.../protocol/IoTDBConfigRegionConnector.java | 10 +-
.../coordinator/plugin/PipePluginCoordinator.java | 29 +
.../pipe/coordinator/task/PipeTaskCoordinator.java | 74 ++-
.../pipe/event/PipeConfigRegionSnapshotEvent.java | 3 +-
.../extractor/ConfigRegionListeningFilter.java | 89 +--
.../pipe/extractor/ConfigRegionListeningQueue.java | 23 +-
.../pipe/extractor/IoTDBConfigRegionExtractor.java | 74 ++-
...ConfigPhysicalPlanTablePatternParseVisitor.java | 135 +++++
...ConfigPhysicalPlanTreePatternParseVisitor.java} | 6 +-
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 170 +++++-
.../PipeConfigPhysicalPlanTSStatusVisitor.java | 76 +++
.../manager/schema/ClusterSchemaManager.java | 158 +++--
.../persistence/executor/ConfigPlanExecutor.java | 5 +
.../confignode/persistence/pipe/PipeInfo.java | 2 +-
.../persistence/pipe/PipePluginInfo.java | 39 +-
.../confignode/persistence/pipe/PipeTaskInfo.java | 28 +-
.../schema/CNPhysicalPlanGenerator.java | 77 ++-
.../persistence/schema/ClusterSchemaInfo.java | 77 +++
.../confignode/persistence/schema/ConfigMTree.java | 29 +-
.../schema/ConfignodeSnapshotParser.java | 30 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 23 +-
.../procedure/env/RegionMaintainHandler.java | 13 +-
.../impl/pipe/task/AlterPipeProcedureV2.java | 4 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 42 +-
.../impl/region/AddRegionPeerProcedure.java | 84 +--
.../impl/region/ReconstructRegionProcedure.java | 209 +++++++
.../impl/region/RegionMigrateProcedure.java | 39 +-
.../impl/region/RegionOperationProcedure.java | 39 +-
.../impl/region/RemoveRegionPeerProcedure.java | 52 +-
.../impl/schema/DeleteDatabaseProcedure.java | 2 +-
.../impl/schema/DeleteTimeSeriesProcedure.java | 4 +-
.../table/AbstractAlterOrDropTableProcedure.java | 10 +-
.../impl/schema/table/AddTableColumnProcedure.java | 16 +-
.../impl/schema/table/CreateTableProcedure.java | 20 +-
.../impl/schema/table/DeleteDevicesProcedure.java | 43 +-
.../schema/table/DropTableColumnProcedure.java | 21 +-
.../impl/schema/table/DropTableProcedure.java | 24 +-
.../schema/table/RenameTableColumnProcedure.java | 18 +-
.../schema/table/SetTablePropertiesProcedure.java | 18 +-
.../impl/trigger/CreateTriggerProcedure.java | 9 +-
.../procedure/state/ReconstructRegionState.java | 10 +-
.../procedure/store/ProcedureFactory.java | 40 +-
.../confignode/procedure/store/ProcedureType.java | 12 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 52 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 40 ++
.../response/pipe/PipePluginTableRespTest.java | 4 +-
.../connector/PipeConfigNodeThriftRequestTest.java | 12 +-
...igPhysicalPlanTablePatternParseVisitorTest.java | 133 ++++
...igPhysicalPlanTreePatternParseVisitorTest.java} | 44 +-
.../pipe/annotation/PipePluginAnnotationTest.java} | 23 +-
.../pipe/receiver/PipeEnrichedProcedureTest.java | 194 ++++++
.../schema/table/AddTableColumnProcedureTest.java | 8 +-
.../schema/table/CreateTableProcedureTest.java | 5 +-
.../schema/table/DeleteDevicesProcedureTest.java | 5 +-
.../schema/table/DropTableColumnProcedureTest.java | 4 +-
.../impl/schema/table/DropTableProcedureTest.java | 4 +-
.../table/RenameTableColumnProcedureTest.java | 4 +-
.../table/SetTablePropertiesProcedureTest.java | 6 +-
.../apache/iotdb/consensus/pipe/PipeConsensus.java | 57 +-
.../apache/iotdb/consensus/ratis/RatisClient.java | 6 +-
.../apache/iotdb/consensus/ratis/utils/Utils.java | 1 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 4 +-
.../dataregion/DataExecutionVisitor.java | 9 +-
.../schemaregion/SchemaExecutionVisitor.java | 15 +-
.../schemaregion/SchemaRegionStateMachine.java | 15 +-
.../LoadAnalyzeException.java} | 8 +-
.../LoadAnalyzeTableColumnDisorderException.java} | 7 +-
.../LoadAnalyzeTypeMismatchException.java} | 6 +-
.../pipe/agent/plugin/PipeDataNodePluginAgent.java | 20 +-
.../dataregion/PipeDataRegionPluginAgent.java | 49 ++
.../agent/task/connection/PipeEventCollector.java | 35 +-
.../client/IoTDBDataNodeAsyncClientManager.java | 2 +-
.../batch/PipeTabletEventTsFileBatch.java | 384 +++---------
.../PipeTransferDataNodeHandshakeV1Req.java | 10 +-
.../evolvable/request/PipeTransferPlanNodeReq.java | 18 +-
.../request/PipeTransferSchemaSnapshotSealReq.java | 94 ++-
.../request/PipeTransferTabletInsertNodeReqV2.java | 2 +-
.../request/PipeTransferTabletRawReq.java | 4 +-
.../request/PipeTransferTabletRawReqV2.java | 9 +-
.../request/PipeTransferTsFileSealWithModReq.java | 42 +-
.../airgap/IoTDBDataNodeAirGapConnector.java | 4 +
.../airgap/IoTDBDataRegionAirGapConnector.java | 4 +
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 19 +-
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 2 +
.../connector/protocol/opcua/OpcUaConnector.java | 4 +
.../connector/protocol/opcua/OpcUaNameSpace.java | 44 +-
.../pipeconsensus/PipeConsensusAsyncConnector.java | 4 +
.../pipeconsensus/PipeConsensusSyncConnector.java | 4 +
.../request/PipeConsensusDeleteNodeReq.java | 4 +-
.../async/IoTDBDataRegionAsyncConnector.java | 14 +-
.../PipeTransferTabletBatchEventHandler.java | 2 +-
.../async/handler/PipeTransferTsFileHandler.java | 13 +-
.../thrift/sync/IoTDBDataNodeSyncConnector.java | 4 +
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 23 +-
.../thrift/sync/IoTDBSchemaRegionConnector.java | 26 +-
.../protocol/websocket/WebSocketConnector.java | 2 +
.../protocol/writeback/WriteBackConnector.java | 199 +++++-
.../util/builder/PipeTableModeTsFileBuilder.java | 273 +++++++++
.../util/builder/PipeTreeModelTsFileBuilder.java | 268 ++++++++
.../connector/util/builder/PipeTsFileBuilder.java | 162 +++++
.../util/{ => cacher}/LeaderCacheUtils.java | 2 +-
.../sorter/PipeTableModelTabletEventSorter.java | 273 +++++++++
.../util/{ => sorter}/PipeTabletEventSorter.java | 107 +---
.../sorter/PipeTreeModelTabletEventSorter.java | 123 ++++
.../pipe/consensus/deletion/DeletionResource.java | 2 +-
.../db/pipe/event/common/PipeInsertionEvent.java | 15 +-
.../common/deletion/PipeDeleteDataNodeEvent.java | 24 +-
.../db/pipe/event/common/row/PipeRowCollector.java | 2 +-
.../schema/PipeSchemaRegionSnapshotEvent.java | 75 ++-
.../schema/PipeSchemaSerializableEventType.java | 15 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 2 +-
.../tablet/parser/TabletInsertionEventParser.java | 8 +-
.../TabletInsertionEventTreePatternParser.java | 13 +-
.../scan/TsFileInsertionEventScanParser.java | 42 +-
.../table/TsFileInsertionEventTableParser.java | 2 +-
.../dataregion/DataRegionListeningFilter.java | 21 +-
.../dataregion/IoTDBDataRegionExtractor.java | 156 ++---
...oricalDataRegionTsFileAndDeletionExtractor.java | 4 +-
.../listener/PipeInsertionDataNodeListener.java | 27 +-
.../schemaregion/IoTDBSchemaRegionExtractor.java | 28 +-
.../PipePlanTablePatternParseVisitor.java | 77 +++
...r.java => PipePlanTreePatternParseVisitor.java} | 7 +-
.../schemaregion/SchemaRegionListeningFilter.java | 87 +--
.../schemaregion/SchemaRegionListeningQueue.java | 8 +-
.../pipe/metric/PipeDataNodeReceiverMetrics.java | 48 +-
.../processor/aggregate/AggregateProcessor.java | 2 +
.../StandardStatisticsOperatorProcessor.java | 2 +
.../processor/TumblingWindowingProcessor.java | 2 +
.../changing/ChangingValueSamplingProcessor.java | 2 +
.../sdt/SwingingDoorTrendingSamplingProcessor.java | 2 +
.../tumbling/TumblingTimeSamplingProcessor.java | 2 +
.../pipeconsensus/PipeConsensusProcessor.java | 4 +
.../schemachange/RenameDatabaseProcessor.java | 11 +-
.../twostage/plugin/TwoStageCountProcessor.java | 37 +-
.../pipeconsensus/PipeConsensusReceiver.java | 33 +-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 223 +++++--
.../statement/PipeConvertedInsertRowStatement.java | 30 +-
.../PipeConvertedInsertTabletStatement.java | 52 +-
.../visitor/PipePlanToStatementVisitor.java | 54 +-
.../PipeStatementTablePatternParseVisitor.java | 33 +-
...a => PipeStatementTreePatternParseVisitor.java} | 6 +-
...leStatementDataTypeConvertExecutionVisitor.java | 10 +-
...eeStatementDataTypeConvertExecutionVisitor.java | 3 +-
...r.java => PipeTreeStatementToBatchVisitor.java} | 2 +-
.../pipe/resource/memory/PipeMemoryWeightUtil.java | 21 +-
.../resource/tsfile/PipeTsFileResourceManager.java | 5 +-
.../iotdb/db/protocol/client/ConfigNodeClient.java | 57 ++
.../impl/DataNodeInternalRPCServiceImpl.java | 2 +
.../execution/executor/RegionWriteExecutor.java | 55 +-
.../function/table/ExcludeColumnFunction.java | 4 +-
.../execution/function/table/HOPTableFunction.java | 1 +
.../operator/process/EnforceSingleRowOperator.java | 5 +-
.../merge/comparator/JoinKeyComparatorFactory.java | 2 +-
.../schema/source/TableDeviceQuerySource.java | 4 +-
.../InformationSchemaContentSupplierFactory.java | 372 ++++++++++--
.../relational/MergeSortSemiJoinOperator.java | 228 +++++++
.../relational/aggregation/AccumulatorFactory.java | 265 +++++++-
.../relational/aggregation/AggregationMask.java | 204 +++++++
.../relational/aggregation/AvgAccumulator.java | 122 +++-
.../relational/aggregation/CountAccumulator.java | 22 +-
...IfAccumulator.java => CountAllAccumulator.java} | 48 +-
.../relational/aggregation/CountIfAccumulator.java | 22 +-
.../relational/aggregation/ExtremeAccumulator.java | 94 ++-
.../relational/aggregation/FirstAccumulator.java | 158 +++--
.../relational/aggregation/FirstByAccumulator.java | 164 +++--
.../aggregation/FirstByDescAccumulator.java | 132 +++-
.../aggregation/FirstDescAccumulator.java | 126 +++-
.../relational/aggregation/LastAccumulator.java | 140 ++++-
.../relational/aggregation/LastByAccumulator.java | 146 ++++-
.../aggregation/LastByDescAccumulator.java | 150 ++++-
.../aggregation/LastDescAccumulator.java | 144 ++++-
...dAccumulator.java => MaskedRecordIterator.java} | 40 +-
.../relational/aggregation/MaxAccumulator.java | 140 ++++-
.../relational/aggregation/MinAccumulator.java | 140 ++++-
.../relational/aggregation/SumAccumulator.java | 114 +++-
.../relational/aggregation/TableAccumulator.java | 2 +-
.../relational/aggregation/TableAggregator.java | 5 +-
.../aggregation/TableMaxMinByBaseAccumulator.java | 146 +++--
.../aggregation/TableModeAccumulator.java | 192 ++++--
.../aggregation/TableVarianceAccumulator.java | 158 +++--
.../UserDefinedAggregateFunctionAccumulator.java | 35 +-
.../aggregation/grouped/GroupedAccumulator.java | 4 +-
.../aggregation/grouped/GroupedAggregator.java | 6 +-
.../aggregation/grouped/GroupedAvgAccumulator.java | 119 +++-
.../grouped/GroupedCountAccumulator.java | 22 +-
.../grouped/GroupedCountIfAccumulator.java | 22 +-
.../grouped/GroupedExtremeAccumulator.java | 95 ++-
.../grouped/GroupedFirstAccumulator.java | 153 ++++-
.../grouped/GroupedFirstByAccumulator.java | 191 ++++--
.../grouped/GroupedLastAccumulator.java | 133 +++-
.../grouped/GroupedLastByAccumulator.java | 150 ++++-
.../aggregation/grouped/GroupedMaxAccumulator.java | 141 ++++-
.../grouped/GroupedMaxMinByBaseAccumulator.java | 147 ++++-
.../aggregation/grouped/GroupedMinAccumulator.java | 141 ++++-
.../grouped/GroupedModeAccumulator.java | 246 ++++++--
.../aggregation/grouped/GroupedSumAccumulator.java | 111 +++-
.../GroupedUserDefinedAggregateAccumulator.java | 14 +-
.../grouped/GroupedVarianceAccumulator.java | 159 +++--
.../aggregation/grouped/hash/GroupByHash.java | 2 +
.../aggregation/grouped/hash/MarkDistinctHash.java | 90 +++
.../relational/ColumnTransformerBuilder.java | 20 +-
.../iotdb/db/queryengine/plan/Coordinator.java | 8 +
.../db/queryengine/plan/analyze/AnalyzeUtils.java | 9 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 30 +-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 27 +-
.../analyze/load/LoadTsFileTableSchemaCache.java | 17 +-
.../load/LoadTsFileToTableModelAnalyzer.java | 34 +-
.../load/LoadTsFileToTreeModelAnalyzer.java | 22 +-
.../analyze/load/LoadTsFileTreeSchemaCache.java | 3 +-
.../load/TreeSchemaAutoCreatorAndVerifier.java | 30 +-
.../execution/config/TableConfigTaskVisitor.java | 141 ++++-
.../execution/config/TreeConfigTaskVisitor.java | 30 +-
.../config/executor/ClusterConfigTaskExecutor.java | 159 ++++-
.../config/executor/IConfigTaskExecutor.java | 19 +-
.../config/metadata/DropPipePluginTask.java | 13 +-
.../config/metadata/ShowPipePluginsTask.java | 20 +-
.../ExtendRegionTask.java} | 19 +-
.../metadata/{ => region}/MigrateRegionTask.java | 4 +-
.../ReconstructRegionTask.java} | 19 +-
.../RemoveRegionTask.java} | 19 +-
.../config/metadata/relational/ShowDBTask.java | 4 +-
.../metadata/relational/ShowTablesDetailsTask.java | 47 +-
.../config/metadata/relational/ShowTablesTask.java | 35 +-
.../execution/config/sys/pipe/AlterPipeTask.java | 8 +-
.../execution/config/sys/pipe/DropPipeTask.java | 7 +-
.../execution/config/sys/pipe/ShowPipeTask.java | 3 +-
.../execution/config/sys/pipe/StartPipeTask.java | 7 +-
.../execution/config/sys/pipe/StopPipeTask.java | 7 +-
.../db/queryengine/plan/parser/ASTVisitor.java | 38 +-
.../plan/planner/LogicalPlanVisitor.java | 7 +-
.../plan/planner/TableOperatorGenerator.java | 95 ++-
.../plan/planner/plan/node/PlanGraphPrinter.java | 12 +
.../plan/planner/plan/node/PlanNodeType.java | 4 +
.../plan/planner/plan/node/PlanVisitor.java | 5 +
.../node/metadata/write/AlterTimeSeriesNode.java | 9 +-
.../node/metadata/write/CreateTimeSeriesNode.java | 74 +--
.../plan/node/pipe/PipeEnrichedDeleteDataNode.java | 66 +-
.../plan/node/pipe/PipeEnrichedInsertNode.java | 44 +-
.../plan/node/pipe/PipeEnrichedWritePlanNode.java | 23 +-
.../relational/analyzer/ExpressionAnalyzer.java | 37 +-
.../plan/relational/analyzer/Scope.java | 8 +-
.../relational/analyzer/StatementAnalyzer.java | 116 ++--
.../plan/relational/metadata/Metadata.java | 4 +-
.../relational/metadata/TableMetadataImpl.java | 53 +-
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 14 +-
.../fetcher/TableHeaderSchemaValidator.java | 11 +-
.../plan/relational/planner/IrTypeAnalyzer.java | 15 +-
.../plan/relational/planner/QueryPlanner.java | 2 +-
.../plan/relational/planner/RelationPlanner.java | 91 +--
.../relational/planner/SimplePlanRewriter.java | 88 +++
.../plan/relational/planner/SubqueryPlanner.java | 84 ++-
.../relational/planner/TableLogicalPlanner.java | 15 +-
.../distribute/TableDistributedPlanGenerator.java | 24 +
.../planner/iterative/rule/PruneApplyColumns.java | 138 +++++
.../iterative/rule/PruneApplyCorrelation.java | 70 +++
.../iterative/rule/PruneApplySourceColumns.java | 95 +++
.../rule/RemoveUnreferencedScalarApplyNodes.java | 42 ++
.../rule/RemoveUnreferencedScalarSubqueries.java | 70 +++
.../TransformFilteringSemiJoinToInnerJoin.java | 150 +++++
...mUncorrelatedInPredicateSubqueryToSemiJoin.java | 95 +++
.../plan/relational/planner/node/Patterns.java | 7 +-
.../plan/relational/planner/node/SemiJoinNode.java | 181 ++++++
.../node/schema/CreateOrUpdateTableDeviceNode.java | 14 +-
.../DataNodeLocationSupplierFactory.java | 11 +
.../optimizations/LogicalOptimizeFactory.java | 26 +-
.../PushAggregationIntoTableScan.java | 4 +
.../optimizations/PushPredicateIntoTableScan.java | 235 ++++++-
...mQuantifiedComparisonApplyToCorrelatedJoin.java | 341 +++++++++++
.../optimizations/UnaliasSymbolReferences.java | 29 +
.../sql/ast/AbstractQueryDeviceWithCache.java | 2 +
.../relational/sql/ast/AbstractTraverseDevice.java | 8 +
.../plan/relational/sql/ast/AlterPipe.java | 22 +-
.../plan/relational/sql/ast/AstVisitor.java | 14 +-
.../sql/ast/{Update.java => CountStatement.java} | 59 +-
.../relational/sql/ast/CreateOrUpdateDevice.java | 31 +-
.../plan/relational/sql/ast/CreatePipe.java | 4 +-
.../plan/relational/sql/ast/Delete.java | 28 +-
.../plan/relational/sql/ast/DeleteDevice.java | 2 -
.../plan/relational/sql/ast/DropColumn.java | 3 -
.../plan/relational/sql/ast/DropPipe.java | 10 +-
.../plan/relational/sql/ast/LoadTsFile.java | 9 +
.../plan/relational/sql/ast/PipeEnriched.java | 8 +-
.../plan/relational/sql/ast/ShowPipes.java | 10 +-
.../plan/relational/sql/ast/StartPipe.java | 8 +-
.../ast/{Statement.java => StartRepairData.java} | 12 +-
.../plan/relational/sql/ast/Statement.java | 4 +-
.../plan/relational/sql/ast/StopPipe.java | 8 +-
.../ast/{Statement.java => StopRepairData.java} | 12 +-
.../plan/relational/sql/ast/Update.java | 6 +-
.../plan/relational/sql/parser/AstBuilder.java | 67 +-
.../plan/relational/sql/rewrite/ShowRewrite.java | 51 +-
.../plan/scheduler/load/LoadTsFileScheduler.java | 9 +-
.../plan/statement/StatementVisitor.java | 18 +-
.../plan/statement/crud/InsertBaseStatement.java | 22 +-
.../plan/statement/crud/InsertRowStatement.java | 18 +-
.../plan/statement/crud/LoadTsFileStatement.java | 9 +
.../metadata/AlterTimeSeriesStatement.java | 7 +-
.../metadata/pipe/AlterPipeStatement.java | 31 +-
.../metadata/pipe/DropPipePluginStatement.java | 17 +-
.../statement/metadata/pipe/DropPipeStatement.java | 23 +-
.../metadata/pipe/ShowPipePluginsStatement.java | 23 +-
.../metadata/pipe/ShowPipesStatement.java | 18 +-
.../metadata/pipe/StartPipeStatement.java | 17 +-
.../statement/metadata/pipe/StopPipeStatement.java | 17 +-
.../ExtendRegionStatement.java} | 32 +-
.../{ => region}/MigrateRegionStatement.java | 2 +-
.../ReconstructRegionStatement.java} | 42 +-
.../RemoveRegionStatement.java} | 32 +-
.../plan/statement/pipe/PipeEnrichedStatement.java | 10 +-
.../attribute/DeviceAttributeStore.java | 10 +-
.../attribute/IDeviceAttributeStore.java | 6 +-
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 136 +++--
.../schemaregion/mtree/impl/mem/MemMTreeStore.java | 52 +-
.../mtree/impl/mem/mnode/basic/BasicMNode.java | 22 +-
.../impl/mem/snapshot/MemMTreeSnapshotUtil.java | 37 +-
.../schemaengine/table/InformationSchemaUtils.java | 11 +-
.../db/service/metrics/DataNodeMetricsHelper.java | 4 +-
.../metrics/memory/ConsensusMemoryMetrics.java | 73 +++
.../metrics/memory/GlobalMemoryMetrics.java | 132 ++++
.../metrics/memory/QueryEngineMemoryMetrics.java | 179 ++++++
.../metrics/memory/SchemaEngineMemoryMetrics.java | 124 ++++
.../metrics/memory/StorageEngineMemoryMetrics.java | 210 +++++++
.../metrics/memory/StreamEngineMemoryMetrics.java | 73 +++
.../metrics/memory/ThresholdMemoryMetrics.java | 194 ------
.../db/storageengine/dataregion/DataRegion.java | 123 ++--
.../task/InsertionCrossSpaceCompactionTask.java | 3 -
.../utils/CompactionTableSchemaCollector.java | 2 +-
.../compaction/repair/RepairDataFileScanUtil.java | 30 +-
.../schedule/CompactionScheduleContext.java | 21 +
.../compaction/schedule/CompactionScheduler.java | 60 +-
.../impl/RewriteCrossSpaceCompactionSelector.java | 11 +-
.../dataregion/memtable/TsFileProcessor.java | 4 +
.../dataregion/modification/ModificationFile.java | 89 ++-
.../modification/PartitionLevelModFileManager.java | 5 +-
.../dataregion/snapshot/SnapshotTaker.java | 2 +-
.../dataregion/tsfile/TsFileResource.java | 10 +-
.../tsfile/timeindex/ArrayDeviceTimeIndex.java | 20 +
.../db/storageengine/load/LoadTsFileManager.java | 23 +-
.../load/active/ActiveLoadDirScanner.java | 27 +-
.../load/active/ActiveLoadTsFileLoader.java | 13 +-
.../LoadConvertedInsertTabletStatement.java | 18 +-
...leStatementDataTypeConvertExecutionVisitor.java | 10 +-
...eeStatementDataTypeConvertExecutionVisitor.java | 3 +-
.../converter/LoadTsFileDataTypeConverter.java | 78 ++-
.../load/splitter/TsFileSplitter.java | 5 +-
.../broker/SubscriptionPrefetchingQueue.java | 2 +
.../db/subscription/event/SubscriptionEvent.java | 21 +-
.../batch/SubscriptionPipeTabletEventBatch.java | 24 +-
.../batch/SubscriptionPipeTsFileEventBatch.java | 11 +-
.../db/tools/schema/SRStatementGenerator.java | 208 +++++--
.../tools/schema/SchemaRegionSnapshotParser.java | 32 +-
.../iotdb/db/utils/constant/SqlConstant.java | 1 +
.../iotdb/db/utils/datastructure/BinaryTVList.java | 6 +-
.../db/utils/datastructure/BooleanTVList.java | 6 +-
.../iotdb/db/utils/datastructure/DoubleTVList.java | 6 +-
.../iotdb/db/utils/datastructure/FloatTVList.java | 6 +-
.../iotdb/db/utils/datastructure/IntTVList.java | 6 +-
.../iotdb/db/utils/datastructure/LongTVList.java | 6 +-
.../connector/PipeDataNodeThriftRequestTest.java | 22 +-
.../PipeStatementTablePatternParseVisitorTest.java | 57 ++
... PipeStatementTreePatternParseVisitorTest.java} | 20 +-
.../pipe/connector/PipeTabletEventSorterTest.java | 263 ++++++--
.../db/pipe/consensus/DeletionResourceTest.java | 39 +-
.../pipe/event/PipeTabletInsertionEventTest.java | 25 +-
.../PipePlanTablePatternParseVisitorTest.java | 129 ++++
...va => PipePlanTreePatternParseVisitorTest.java} | 32 +-
.../annotation/PipePluginAnnotationTest.java} | 23 +-
.../resource/PipeTsFileResourceManagerTest.java | 2 +-
.../metadata/write/MetadataWriteNodeSerDeTest.java | 2 +-
.../plan/relational/analyzer/DistinctTest.java | 91 +++
.../plan/relational/planner/SubqueryTest.java | 310 ++++++++++
.../planner/assertions/PlanMatchPattern.java | 17 +
.../planner/assertions/SemiJoinMatcher.java | 79 +++
.../storageengine/buffer/BloomFilterCacheTest.java | 5 +-
.../storageengine/dataregion/DataRegionTest.java | 24 +-
.../compaction/CompactionValidationTest.java | 7 +-
.../InsertionCrossSpaceCompactionSelectorTest.java | 4 +-
.../cross/InsertionCrossSpaceCompactionTest.java | 64 +-
.../inner/InnerSpaceCompactionSelectorTest.java | 2 +-
.../repair/RepairDataFileScanUtilTest.java | 2 +-
.../TableModelFastCompactionPerformerTest.java | 13 +-
...TableModelReadChunkCompactionPerformerTest.java | 13 +-
...TableModelReadPointCompactionPerformerTest.java | 13 +-
.../modification/ModificationFileTest.java | 90 ++-
.../reader/chunk/MemAlignedChunkLoaderTest.java | 2 +-
.../read/reader/chunk/MemChunkLoaderTest.java | 12 +-
.../file/UnsealedTsFileRecoverPerformerTest.java | 4 +-
.../iotdb/db/tools/TsFileAndModSettleToolTest.java | 2 +-
.../iotdb/db/tools/TsFileSelfCheckToolTest.java | 6 +-
.../iotdb/db/tools/TsFileSketchToolTest.java | 10 +-
.../db/utils/SchemaRegionSnapshotParserTest.java | 200 ++++--
iotdb-core/node-commons/pom.xml | 4 +
.../conf/iotdb-system.properties.template | 6 +-
.../exception}/table/ColumnNotExistsException.java | 2 +-
.../table/TableAlreadyExistsException.java | 2 +-
.../exception}/table/TableNotExistsException.java | 2 +-
.../commons/pipe/agent/plugin/PipePluginAgent.java | 17 +-
.../builtin/connector/PlaceholderConnector.java | 4 +
.../connector/donothing/DoNothingConnector.java | 4 +
.../iotdb/airgap/IoTDBAirGapConnector.java | 4 +
.../consensus/PipeConsensusAsyncConnector.java | 4 +
.../iotdb/thrift/IoTDBLegacyPipeConnector.java | 2 +
.../iotdb/thrift/IoTDBThriftAsyncConnector.java | 5 +
.../iotdb/thrift/IoTDBThriftConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSslConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSyncConnector.java | 5 +
.../builtin/connector/opcua/OpcUaConnector.java | 4 +
.../connector/websocket/WebSocketConnector.java | 2 +
.../connector/writeback/WriteBackConnector.java | 4 +
.../extractor/donothing/DoNothingExtractor.java | 4 +
.../builtin/extractor/iotdb/IoTDBExtractor.java | 4 +
.../processor/donothing/DoNothingProcessor.java | 4 +
.../throwing/ThrowingExceptionProcessor.java | 4 +
.../agent/plugin/meta/PipePluginMetaKeeper.java | 47 +-
.../commons/pipe/agent/task/PipeTaskAgent.java | 2 +-
.../pipe/agent/task/meta/PipeMetaKeeper.java | 12 +-
.../pipe/agent/task/meta/PipeStaticMeta.java | 14 +
.../config/constant/PipeExtractorConstant.java | 3 +
.../connector/client/IoTDBSyncClientManager.java | 28 +-
.../payload/thrift/request/PipeRequestType.java | 4 +-
.../thrift/request/PipeTransferFileSealReqV2.java | 13 +-
.../connector/protocol/IoTDBAirGapConnector.java | 14 +-
.../pipe/connector/protocol/IoTDBConnector.java | 10 +
.../connector/protocol/IoTDBSslSyncConnector.java | 4 +
.../options/PipeInclusionOptions.java | 43 +-
.../pipe/datastructure/pattern/TablePattern.java | 29 +-
.../pipe/datastructure/pattern/TreePattern.java | 25 +-
.../{ => resource}/PersistentResource.java | 2 +-
.../commons/pipe/datastructure/result/Result.java} | 37 +-
.../pipe/datastructure/visibility/Visibility.java | 9 +-
.../visibility/VisibilityTestUtils.java | 135 +++++
.../datastructure/visibility/VisibilityUtils.java | 133 ++++
.../commons/pipe/extractor/IoTDBExtractor.java | 103 +++-
.../extractor/IoTDBNonDataRegionExtractor.java | 24 +-
.../commons/pipe/receiver/IoTDBFileReceiver.java | 49 +-
.../commons/pipe/receiver/IoTDBReceiverAgent.java | 7 +-
.../snapshot/PipeSnapshotResourceManager.java | 9 +-
.../apache/iotdb/commons/schema/node/IMNode.java | 18 +-
.../commons/schema/node/visitor/MNodeVisitor.java | 6 +-
.../iotdb/commons/service/metric/enums/Metric.java | 3 +-
.../iotdb/commons/udf/access/RecordIterator.java | 9 +-
.../TableBuiltinAggregationFunction.java | 2 +
.../org/apache/iotdb/commons/utils/FileUtils.java | 49 ++
.../org/apache/iotdb/commons/utils/PathUtils.java | 2 +-
.../apache/iotdb/commons/utils/RetryUtils.java} | 30 +-
.../annotation/PipePluginAnnotationTest.java} | 23 +-
.../db/relational/grammar/sql/RelationalSql.g4 | 11 +-
.../thrift-commons/src/main/thrift/common.thrift | 3 +-
.../src/main/thrift/confignode.thrift | 69 +++
pom.xml | 6 +-
608 files changed, 23175 insertions(+), 6095 deletions(-)
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
index 2b3998365a5,00000000000..e01428e4736
mode 100644,000000..100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/ExcludeColumnFunction.java
@@@ -1,105 -1,0 +1,103 @@@
+/*
+ * 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.queryengine.execution.function.table;
+
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class ExcludeColumnFunction implements TableFunction {
+ private final String TBL_PARAM = "DATA";
+ private final String COL_PARAM = "EXCLUDE";
+
+ @Override
+ public List<ParameterSpecification> getArgumentsSpecification() {
+ return Arrays.asList(
+ TableParameterSpecification.builder().name(TBL_PARAM).rowSemantics().build(),
+ ScalarParameterSpecification.builder().name(COL_PARAM).type(Type.STRING).build());
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException {
+ TableArgument tableArgument = (TableArgument) arguments.get(TBL_PARAM);
+ if (tableArgument == null) {
+ throw new UDFException("Table argument is missing");
+ }
+ String excludeColumn = (String) ((ScalarArgument) arguments.get(COL_PARAM)).getValue();
+ ImmutableList.Builder<Integer> requiredColumns = ImmutableList.builder();
+ DescribedSchema.Builder schemaBuilder = DescribedSchema.builder();
+ for (int i = 0; i < tableArgument.getFieldNames().size(); i++) {
+ Optional<String> fieldName = tableArgument.getFieldNames().get(i);
+ if (!fieldName.isPresent() || !fieldName.get().equalsIgnoreCase(excludeColumn)) {
+ requiredColumns.add(i);
- schemaBuilder.addField(
- fieldName,
- tableArgument.getFieldTypes().get(i));
++ schemaBuilder.addField(fieldName, tableArgument.getFieldTypes().get(i));
+ }
+ }
+ return TableFunctionAnalysis.builder()
+ .properColumnSchema(schemaBuilder.build())
+ .requiredColumns(TBL_PARAM, requiredColumns.build())
+ .build();
+ }
+
+ @Override
+ public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new TableFunctionDataProcessor() {
+ @Override
+ public void process(Record input, List<ColumnBuilder> columnBuilders) {
+ for (int i = 0; i < input.size(); i++) {
+ if (input.isNull(i)) {
+ columnBuilders.get(i).appendNull();
+ } else {
+ columnBuilders.get(i).writeObject(input.getObject(i));
+ }
+ }
+ }
+
+ @Override
+ public void finish(List<ColumnBuilder> columnBuilders) {
+ // do nothing
+ }
+ };
+ }
+ };
+ }
+}
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
index 6bf8df15f04,00000000000..4da85a53f12
mode 100644,000000..100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/function/table/HOPTableFunction.java
@@@ -1,176 -1,0 +1,177 @@@
+/*
+ * 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.queryengine.execution.function.table;
+
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+import java.util.Arrays;
++import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class HOPTableFunction implements TableFunction {
+
+ private static final String DATA_PARAMETER_NAME = "DATA";
+ private static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
+ private static final String SLIDE_PARAMETER_NAME = "SLIDE";
+ private static final String SIZE_PARAMETER_NAME = "SIZE";
+ private static final String START_PARAMETER_NAME = "START";
+
+ @Override
+ public List<ParameterSpecification> getArgumentsSpecification() {
+ return Arrays.asList(
+ TableParameterSpecification.builder()
+ .name(DATA_PARAMETER_NAME)
+ .passThroughColumns()
+ .keepWhenEmpty()
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(TIMECOL_PARAMETER_NAME)
+ .type(Type.STRING)
+ .build(),
+ ScalarParameterSpecification.builder().name(SLIDE_PARAMETER_NAME).type(Type.INT64).build(),
+ ScalarParameterSpecification.builder().name(SIZE_PARAMETER_NAME).type(Type.INT64).build(),
+ ScalarParameterSpecification.builder()
+ .name(START_PARAMETER_NAME)
+ .type(Type.TIMESTAMP)
+ .defaultValue(Long.MIN_VALUE)
+ .build());
+ }
+
+ private int findTimeColumnIndex(TableArgument tableArgument, String expectedFieldName) {
+ int requiredIndex = -1;
+ for (int i = 0; i < tableArgument.getFieldTypes().size(); i++) {
+ Optional<String> fieldName = tableArgument.getFieldNames().get(i);
+ if (fieldName.isPresent() && expectedFieldName.equalsIgnoreCase(fieldName.get())) {
+ requiredIndex = i;
+ break;
+ }
+ }
+ return requiredIndex;
+ }
+
+ // TODO: ImmutableMap
+ @Override
+ public TableFunctionAnalysis analyze(Map<String, Argument> arguments) {
+ TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
+ String expectedFieldName =
+ (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
+ int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName);
+ if (requiredIndex == -1) {
+ throw new UDFException("The required field is not found in the input table");
+ }
+ DescribedSchema properColumnSchema =
+ new DescribedSchema.Builder()
+ .addField("window_start", Type.TIMESTAMP)
+ .addField("window_end", Type.TIMESTAMP)
+ .build();
+
+ // outputColumnSchema
+ return TableFunctionAnalysis.builder()
+ .properColumnSchema(properColumnSchema)
+ .requiredColumns(
+ DATA_PARAMETER_NAME,
+ IntStream.range(0, tableArgument.getFieldTypes().size())
+ .boxed()
+ .collect(Collectors.toList()))
+ .build();
+ }
+
+ @Override
+ public TableFunctionProcessorProvider getProcessorProvider(Map<String, Argument> arguments) {
+ TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME);
+ String expectedFieldName =
+ (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
+ int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName);
+
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new HOPDataProcessor(
+ (Long) ((ScalarArgument) arguments.get(START_PARAMETER_NAME)).getValue(),
+ (Long) ((ScalarArgument) arguments.get(SLIDE_PARAMETER_NAME)).getValue(),
+ (Long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue(),
+ requiredIndex);
+ }
+ };
+ }
+
+ private static class HOPDataProcessor implements TableFunctionDataProcessor {
+
+ private final long slide;
+ private final long size;
+ private final int timeColumnIndex;
+ private long curTime;
+
+ public HOPDataProcessor(long startTime, long slide, long size, int timeColumnIndex) {
+ this.slide = slide;
+ this.size = size;
+ this.curTime = startTime;
+ this.timeColumnIndex = timeColumnIndex;
+ }
+
+ @Override
+ public void process(Record input, List<ColumnBuilder> columnBuilders) {
+ long timeValue = input.getLong(timeColumnIndex);
+ if (curTime == Long.MIN_VALUE) {
+ curTime = timeValue;
+ }
+ if (curTime + size <= timeValue) {
+ // jump to appropriate window
+ long move = (timeValue - curTime - size) / slide + 1;
+ curTime += move * slide;
+ }
+ long slideTime = curTime;
+ while (slideTime <= timeValue && slideTime + size > timeValue) {
+ for (int i = 0; i < input.size(); i++) {
+ if (input.isNull(i)) {
+ columnBuilders.get(i + 2).appendNull();
+ } else {
+ columnBuilders.get(i + 2).writeObject(input.getObject(i));
+ }
+ }
+ columnBuilders.get(0).writeLong(slideTime);
+ columnBuilders.get(1).writeLong(slideTime + size);
+ slideTime += slide;
+ }
+ }
+
+ @Override
+ public void finish(List<ColumnBuilder> columnBuilders) {}
+ }
+}
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
index 95df45ae6b6,dbf74b711ca..3e0cabe0422
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
@@@ -74,18 -76,18 +76,22 @@@ public class GroupedUserDefinedAggregat
}
@Override
- public void addInput(int[] groupIds, Column[] arguments) {
+ public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
RecordIterator iterator =
- new RecordIterator(
- Arrays.asList(arguments), inputDataTypes, arguments[0].getPositionCount());
+ mask.isSelectAll()
+ ? new RecordIterator(
+ Arrays.asList(arguments), inputDataTypes, arguments[0].getPositionCount())
+ : new MaskedRecordIterator(Arrays.asList(arguments), inputDataTypes, mask);
+ int[] selectedPositions = mask.getSelectedPositions();
int index = 0;
while (iterator.hasNext()) {
- int groupId = groupIds[index++];
+ int groupId = groupIds[selectedPositions[index]];
+ index++;
State state = getOrCreateState(groupId);
+ if (state == null) {
+ state = aggregateFunction.createState();
+ stateArray.set(groupId, state);
+ }
aggregateFunction.addInput(state, iterator.next());
}
}
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 01377a469ff,e94136cf74c..f28c338b6ac
--- 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
@@@ -145,10 -145,9 +147,11 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SemiJoinNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionProcessorNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
@@@ -220,8 -217,8 +223,9 @@@ import java.util.function.BiFunction
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
+ import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.Iterables.getOnlyElement;
import static java.util.Objects.requireNonNull;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.FIELD;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TIME;
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index e7c3c1de1bf,442d59f8be3..441b31b2c42
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
@@@ -72,7 -72,7 +72,8 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
+ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SemiJoinNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionProcessorNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
@@@ -942,72 -935,18 +943,83 @@@ public class PlanGraphPrinter extends P
}
return render(node, boxValue, context);
}
+
+ @Override
+ public List<String> visitSemiJoin(SemiJoinNode node, GraphContext context) {
+ List<String> boxValue = new ArrayList<>();
+ boxValue.add(String.format("SemiJoin-%s", node.getPlanNodeId().getId()));
+ boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
+ boxValue.add(String.format("SourceJoinSymbol: %s", node.getSourceJoinSymbol()));
+ boxValue.add(
+ String.format("FilteringSourceJoinSymbol: %s", node.getFilteringSourceJoinSymbol()));
+ return render(node, boxValue, context);
+ }
+ @Override
+ public List<String> visitTableFunctionProcessor(
+ TableFunctionProcessorNode node, GraphContext context) {
+ List<String> boxValue = new ArrayList<>();
+ boxValue.add(String.format("TableFunctionProcessor-%s", node.getPlanNodeId().getId()));
+ boxValue.add(String.format("Function: %s", node.getName()));
+ boxValue.add(String.format("ProperOutputs: %s", node.getProperOutputs()));
+ boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
+ boxValue.add(String.format("RequiredSymbols: %s", node.getRequiredSymbols()));
+ if (node.isPruneWhenEmpty()) {
+ boxValue.add("Prune when empty");
+ } else {
+ boxValue.add("Keep when empty");
+ }
+ node.getDataOrganizationSpecification()
+ .ifPresent(
+ specification -> {
+ if (!specification.getPartitionBy().isEmpty()) {
+ boxValue.add(
+ "Partition by: [" + Joiner.on(", ").join(specification.getPartitionBy()) + "]");
+ }
+ specification
+ .getOrderingScheme()
+ .ifPresent(orderingScheme -> boxValue.add("Order by: " + orderingScheme));
+ });
+ if (!node.getArguments().isEmpty()) {
+ node.getArguments().forEach((key, value) -> boxValue.add(formatArgument(key, value)));
+ }
+ return render(node, boxValue, context);
+ }
+
+ private String formatArgument(String argumentName, Argument argument) {
+ if (argument instanceof ScalarArgument) {
+ return formatScalarArgument(argumentName, (ScalarArgument) argument);
+ }
+ if (argument instanceof DescriptorArgument) {
+ return formatDescriptorArgument(argumentName, (DescriptorArgument) argument);
+ } else {
+ return formatTableArgument(argumentName, (TableArgument) argument);
+ }
+ }
+
+ private String formatScalarArgument(String argumentName, ScalarArgument argument) {
+ return format(
+ "%s => ScalarArgument{type=%s, value=%s}",
+ argumentName, argument.getType(), argument.getValue());
+ }
+
+ private String formatDescriptorArgument(String argumentName, DescriptorArgument argument) {
+ String descriptor;
+ if (argument.getDescriptor().isPresent()) {
+ descriptor =
+ argument.getDescriptor().get().getFields().stream()
+ .map(field -> field.getName() + field.getType().map(type -> " " + type).orElse(""))
+ .collect(joining(", ", "(", ")"));
+ } else {
+ descriptor = "NULL";
+ }
+ return format("%s => DescriptorArgument{%s}", argumentName, descriptor);
+ }
+
+ private String formatTableArgument(String argumentName, TableArgument argument) {
+ return format("%s => TableArgument", argumentName);
+ }
+
private String printRegion(TRegionReplicaSet regionReplicaSet) {
return String.format(
"Partition: %s",
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
index 99db38ae149,d04c57e77fb..69e948a3fa0
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
@@@ -127,8 -127,7 +127,9 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
+ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SemiJoinNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionProcessorNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index bbdb55815ac,65ae1c14835..d79fcca4476
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@@ -41,8 -36,8 +41,9 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
+ import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.IrExpressionInterpreter;
import org.apache.iotdb.db.queryengine.plan.relational.planner.PlannerContext;
import org.apache.iotdb.db.queryengine.plan.relational.planner.ScopeAware;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
index 2a556889336,607b1c37135..55f36ec4e89
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
@@@ -52,12 -52,12 +55,13 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignature;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.utils.constant.SqlConstant;
- import org.apache.iotdb.udf.api.customizer.config.AggregateFunctionConfig;
- import org.apache.iotdb.udf.api.customizer.config.ScalarFunctionConfig;
- import org.apache.iotdb.udf.api.customizer.parameter.FunctionParameters;
+ import org.apache.iotdb.rpc.TSStatusCode;
+ import org.apache.iotdb.udf.api.customizer.analysis.AggregateFunctionAnalysis;
+ import org.apache.iotdb.udf.api.customizer.analysis.ScalarFunctionAnalysis;
+ import org.apache.iotdb.udf.api.customizer.parameter.FunctionArguments;
import org.apache.iotdb.udf.api.relational.AggregateFunction;
import org.apache.iotdb.udf.api.relational.ScalarFunction;
+import org.apache.iotdb.udf.api.relational.TableFunction;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.type.BlobType;
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index c987c94268c,a90c07eaaf8..637c7abc7c0
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@@ -57,9 -57,9 +57,10 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SemiJoinNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionProcessorNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
index 058bad7cb1b,2b414438983..c6fd7862bea
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
@@@ -46,9 -45,8 +46,10 @@@ import org.apache.iotdb.db.queryengine.
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SemiJoinNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableFunctionProcessorNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
@@@ -735,114 -733,33 +736,142 @@@ public class UnaliasSymbolReferences im
outputMapping);
}
+ @Override
+ public PlanAndMappings visitSemiJoin(SemiJoinNode node, UnaliasContext context) {
+ // it is assumed that symbols are distinct between SemiJoin source and filtering source. Only
+ // symbols from outer correlation might be the exception
+ PlanAndMappings rewrittenSource = node.getSource().accept(this, context);
+ PlanAndMappings rewrittenFilteringSource = node.getFilteringSource().accept(this, context);
+
+ Map<Symbol, Symbol> outputMapping = new HashMap<>();
+ outputMapping.putAll(rewrittenSource.getMappings());
+ outputMapping.putAll(rewrittenFilteringSource.getMappings());
+
+ SymbolMapper mapper = symbolMapper(outputMapping);
+
+ Symbol newSourceJoinSymbol = mapper.map(node.getSourceJoinSymbol());
+ Symbol newFilteringSourceJoinSymbol = mapper.map(node.getFilteringSourceJoinSymbol());
+ Symbol newSemiJoinOutput = mapper.map(node.getSemiJoinOutput());
+
+ return new PlanAndMappings(
+ new SemiJoinNode(
+ node.getPlanNodeId(),
+ rewrittenSource.getRoot(),
+ rewrittenFilteringSource.getRoot(),
+ newSourceJoinSymbol,
+ newFilteringSourceJoinSymbol,
+ newSemiJoinOutput),
+ outputMapping);
+ }
++
+ @Override
+ public PlanAndMappings visitTableFunction(TableFunctionNode node, UnaliasContext context) {
+ Map<Symbol, Symbol> mapping = new HashMap<>(context.getCorrelationMapping());
+ SymbolMapper mapper = symbolMapper(mapping);
+
+ List<Symbol> newProperOutputs = mapper.map(node.getProperOutputs());
+
+ ImmutableList.Builder<PlanNode> newSources = ImmutableList.builder();
+ ImmutableList.Builder<TableFunctionNode.TableArgumentProperties> newTableArgumentProperties =
+ ImmutableList.builder();
+
+ for (int i = 0; i < node.getChildren().size(); i++) {
+ PlanAndMappings newSource = node.getChildren().get(i).accept(this, context);
+ newSources.add(newSource.getRoot());
+
+ SymbolMapper inputMapper = symbolMapper(new HashMap<>(newSource.getMappings()));
+ TableFunctionNode.TableArgumentProperties properties =
+ node.getTableArgumentProperties().get(i);
+ Optional<DataOrganizationSpecification> newSpecification =
+ properties.getDataOrganizationSpecification().map(inputMapper::mapAndDistinct);
+ TableFunctionNode.PassThroughSpecification newPassThroughSpecification =
+ new TableFunctionNode.PassThroughSpecification(
+ properties.getPassThroughSpecification().isDeclaredAsPassThrough(),
+ properties.getPassThroughSpecification().getColumns().stream()
+ .map(
+ column ->
+ new TableFunctionNode.PassThroughColumn(
+ inputMapper.map(column.getSymbol()), column.isPartitioningColumn()))
+ .collect(toImmutableList()));
+ newTableArgumentProperties.add(
+ new TableFunctionNode.TableArgumentProperties(
+ properties.getArgumentName(),
+ properties.isRowSemantics(),
+ properties.isPruneWhenEmpty(),
+ newPassThroughSpecification,
+ inputMapper.map(properties.getRequiredColumns()),
+ newSpecification));
+ }
+
+ return new PlanAndMappings(
+ new TableFunctionNode(
+ node.getPlanNodeId(),
+ node.getName(),
+ node.getArguments(),
+ newProperOutputs,
+ newSources.build(),
+ newTableArgumentProperties.build()),
+ mapping);
+ }
+
+ @Override
+ public PlanAndMappings visitTableFunctionProcessor(
+ TableFunctionProcessorNode node, UnaliasContext context) {
+ if (node.getChildren().isEmpty()) {
+ Map<Symbol, Symbol> mapping = new HashMap<>(context.getCorrelationMapping());
+ SymbolMapper mapper = symbolMapper(mapping);
+ return new PlanAndMappings(
+ new TableFunctionProcessorNode(
+ node.getPlanNodeId(),
+ node.getName(),
+ mapper.map(node.getProperOutputs()),
+ Optional.empty(),
+ node.isPruneWhenEmpty(),
+ ImmutableList.of(),
+ ImmutableList.of(),
+ Optional.empty(),
+ node.getArguments()),
+ mapping);
+ }
+
+ PlanAndMappings rewrittenSource = node.getChild().accept(this, context);
+ Map<Symbol, Symbol> mapping = new HashMap<>(rewrittenSource.getMappings());
+ SymbolMapper mapper = symbolMapper(mapping);
+
+ List<TableFunctionNode.PassThroughSpecification> newPassThroughSpecification =
+ node.getPassThroughSpecifications().stream()
+ .map(
+ passThroughSpecification ->
+ new TableFunctionNode.PassThroughSpecification(
+ passThroughSpecification.isDeclaredAsPassThrough(),
+ passThroughSpecification.getColumns().stream()
+ .map(
+ column ->
+ new TableFunctionNode.PassThroughColumn(
+ mapper.map(column.getSymbol()),
+ column.isPartitioningColumn()))
+ .collect(toImmutableList())))
+ .collect(Collectors.toList());
+ List<List<Symbol>> newRequiredSymbols =
+ node.getRequiredSymbols().stream().map(mapper::map).collect(toImmutableList());
+
+ Optional<DataOrganizationSpecification> newSpecification =
+ node.getDataOrganizationSpecification().map(mapper::mapAndDistinct);
+
+ TableFunctionProcessorNode rewrittenTableFunctionProcessor =
+ new TableFunctionProcessorNode(
+ node.getPlanNodeId(),
+ node.getName(),
+ mapper.map(node.getProperOutputs()),
+ Optional.of(rewrittenSource.getRoot()),
+ node.isPruneWhenEmpty(),
+ newPassThroughSpecification,
+ newRequiredSymbols,
+ newSpecification,
+ node.getArguments());
+
+ return new PlanAndMappings(rewrittenTableFunctionProcessor, mapping);
+ }
}
private static class UnaliasContext {