This is an automated email from the ASF dual-hosted git repository. yjhjstz pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit d2ee2fc1bd4213b17d48e1cb0f33f2b50523a292 Author: Jianghua Yang <[email protected]> AuthorDate: Sun Oct 12 00:37:44 2025 +0000 orca: implement intra-segment parallel table scan support Add comprehensive parallel table scan capability to GPORCA optimizer, enabling worker-level parallelism within segments for improved query performance on large table scans. Key components: - New CPhysicalParallelTableScan operator and CDistributionSpecWorkerRandom distribution specification for worker-level data distribution - CXformGet2ParallelTableScan transformation with parallel safety checks (excludes CTEs, dynamic scans, foreign tables, replicated tables, etc.) - Cost model integration with parallel_setup_cost and efficiency degradation scaling (logarithmic based on worker count) - DXL serialization/deserialization for CDXLPhysicalParallelTableScan - Plan translation to PostgreSQL SeqScan nodes with parallel_aware=true - Rewindability constraints (parallel scans are non-rewindable) - GUC integration: max_parallel_workers_per_gather controls worker count --- .github/workflows/build-cloudberry.yml | 3 + src/backend/gpopt/config/CConfigParamMapping.cpp | 2 + src/backend/gpopt/gpdbwrappers.cpp | 52 ++++ .../gpopt/translate/CTranslatorDXLToPlStmt.cpp | 222 +++++++++++++- .../gpopt/translate/CTranslatorRelcacheToDXL.cpp | 22 +- src/backend/gpopt/utils/COptTasks.cpp | 8 +- .../libgpdbcost/include/gpdbcost/CCostModelGPDB.h | 9 + .../gporca/libgpdbcost/src/CCostModelGPDB.cpp | 128 +++++++- .../include/gpopt/base/CDistributionSpec.h | 1 + .../gpopt/base/CDistributionSpecWorkerRandom.h | 139 +++++++++ .../include/gpopt/base/CRewindabilitySpec.h | 1 + .../libgpopt/include/gpopt/operators/COperator.h | 1 + .../gpopt/operators/CPhysicalParallelTableScan.h | 140 +++++++++ .../include/gpopt/operators/CPhysicalTableScan.h | 3 +- .../include/gpopt/optimizer/COptimizerConfig.h | 13 +- .../gporca/libgpopt/include/gpopt/search/CGroup.h | 14 + .../libgpopt/include/gpopt/search/CGroupProxy.h | 8 + .../gporca/libgpopt/include/gpopt/xforms/CXform.h | 1 + .../gpopt/xforms/CXformGet2ParallelTableScan.h | 88 ++++++ .../gporca/libgpopt/include/gpopt/xforms/xforms.h | 1 + .../src/base/CDistributionSpecWorkerRandom.cpp | 330 +++++++++++++++++++++ src/backend/gporca/libgpopt/src/base/Makefile | 1 + .../gporca/libgpopt/src/operators/CLogicalGet.cpp | 2 + .../src/operators/CPhysicalCTEProducer.cpp | 1 + .../src/operators/CPhysicalParallelTableScan.cpp | 284 ++++++++++++++++++ src/backend/gporca/libgpopt/src/operators/Makefile | 1 + .../libgpopt/src/optimizer/COptimizerConfig.cpp | 10 +- src/backend/gporca/libgpopt/src/search/CGroup.cpp | 20 ++ src/backend/gporca/libgpopt/src/search/CMemo.cpp | 1 + .../src/translate/CTranslatorExprToDXL.cpp | 14 + .../src/translate/CTranslatorExprToDXLUtils.cpp | 62 +++- .../gporca/libgpopt/src/xforms/CXformFactory.cpp | 1 + .../src/xforms/CXformGet2ParallelTableScan.cpp | 243 +++++++++++++++ .../libgpopt/src/xforms/CXformGet2TableScan.cpp | 11 + src/backend/gporca/libgpopt/src/xforms/Makefile | 1 + .../include/naucrates/dxl/operators/CDXLOperator.h | 1 + .../naucrates/dxl/operators/CDXLOperatorFactory.h | 4 + .../dxl/operators/CDXLPhysicalParallelTableScan.h | 104 +++++++ .../dxl/operators/CDXLPhysicalTableScan.h | 3 +- .../include/naucrates/dxl/xml/dxltokens.h | 2 + .../include/naucrates/md/CMDRelationCtasGPDB.h | 6 + .../include/naucrates/md/CMDRelationGPDB.h | 18 +- .../include/naucrates/md/IMDRelation.h | 6 + .../include/naucrates/md/IMDRelationCtas.h | 7 + .../gporca/libnaucrates/src/md/CMDRelationGPDB.cpp | 44 ++- .../src/operators/CDXLOperatorFactory.cpp | 26 ++ .../operators/CDXLPhysicalParallelTableScan.cpp | 162 ++++++++++ .../gporca/libnaucrates/src/operators/Makefile | 1 + .../src/parser/CParseHandlerOptimizerConfig.cpp | 2 +- .../gporca/libnaucrates/src/xml/dxltokens.cpp | 2 + src/backend/gporca/server/src/startup/main.cpp | 18 ++ .../gporca/server/src/unittest/CTestUtils.cpp | 4 +- .../dxl/statistics/CCardinalityTestUtils.cpp | 3 +- .../src/unittest/dxl/statistics/CHistogramTest.cpp | 3 +- .../unittest/dxl/statistics/CStatisticsTest.cpp | 2 +- .../src/unittest/gpopt/minidump/CICGTest.cpp | 2 +- .../unittest/gpopt/minidump/CMiniDumperDXLTest.cpp | 2 +- .../unittest/gpopt/minidump/CMissingStatsTest.cpp | 2 +- .../src/unittest/gpopt/xforms/CXformTest.cpp | 2 +- src/backend/optimizer/plan/planner.c | 21 ++ src/include/gpopt/gpdbwrappers.h | 5 + .../gpopt/translate/CTranslatorDXLToPlStmt.h | 11 + src/include/gpopt/utils/COptTasks.h | 6 +- src/include/optimizer/orcaopt.h | 1 + src/test/regress/GNUmakefile | 6 + src/test/regress/excluded_tests.conf | 14 + src/test/regress/expected/brin.out | 1 + src/test/regress/expected/brin_bloom.out | 1 + src/test/regress/expected/brin_bloom_optimizer.out | 1 + src/test/regress/expected/brin_multi.out | 1 + src/test/regress/expected/brin_multi_optimizer.out | 1 + .../regress/expected/brin_multi_optimizer_1.out | 1 + src/test/regress/expected/brin_optimizer.out | 1 + src/test/regress/expected/create_index.out | 2 +- .../regress/expected/create_index_optimizer.out | 2 +- src/test/regress/expected/rpt_optimizer.out | 15 +- src/test/regress/expected/tuplesort.out | 2 +- src/test/regress/expected/tuplesort_optimizer.out | 2 +- .../regress/expected/workfile/hashjoin_spill.out | 2 + src/test/regress/sql/brin.sql | 1 + src/test/regress/sql/brin_bloom.sql | 1 + src/test/regress/sql/brin_multi.sql | 1 + src/test/regress/sql/create_index.sql | 2 +- src/test/regress/sql/tuplesort.sql | 2 +- src/test/regress/sql/workfile/hashjoin_spill.sql | 4 + 85 files changed, 2323 insertions(+), 48 deletions(-) diff --git a/.github/workflows/build-cloudberry.yml b/.github/workflows/build-cloudberry.yml index b0191a2d0e9..c00dcde0486 100644 --- a/.github/workflows/build-cloudberry.yml +++ b/.github/workflows/build-cloudberry.yml @@ -342,6 +342,9 @@ jobs: }, {"test":"ic-cbdb-parallel", "make_configs":["src/test/regress:installcheck-cbdb-parallel"] + }, + {"test":"ic-orca-parallel", + "make_configs":["src/test/regress:installcheck-orca-parallel"] } ] }' diff --git a/src/backend/gpopt/config/CConfigParamMapping.cpp b/src/backend/gpopt/config/CConfigParamMapping.cpp index 62255364ba8..5bd9091f604 100644 --- a/src/backend/gpopt/config/CConfigParamMapping.cpp +++ b/src/backend/gpopt/config/CConfigParamMapping.cpp @@ -454,6 +454,8 @@ CConfigParamMapping::PackConfigParamInBitset( // disable table scan if the corresponding GUC is turned off traceflag_bitset->ExchangeSet( GPOPT_DISABLE_XFORM_TF(CXform::ExfGet2TableScan)); + traceflag_bitset->ExchangeSet( + GPOPT_DISABLE_XFORM_TF(CXform::ExfGet2ParallelTableScan)); } if (!optimizer_enable_push_join_below_union_all) diff --git a/src/backend/gpopt/gpdbwrappers.cpp b/src/backend/gpopt/gpdbwrappers.cpp index 4d8d8c59100..97b149d6164 100644 --- a/src/backend/gpopt/gpdbwrappers.cpp +++ b/src/backend/gpopt/gpdbwrappers.cpp @@ -25,6 +25,8 @@ #include <limits> // std::numeric_limits #include "gpos/base.h" +#include "gpopt/base/COptCtxt.h" +#include "gpopt/optimizer/COptimizerConfig.h" #include "gpos/error/CAutoExceptionStack.h" #include "gpos/error/CException.h" @@ -36,8 +38,10 @@ extern "C" { #include "access/amapi.h" #include "access/external.h" #include "access/genam.h" +#include "access/parallel.h" #include "catalog/pg_aggregate.h" #include "catalog/pg_inherits.h" +#include "cdb/cdbvars.h" #include "foreign/fdwapi.h" #include "nodes/nodeFuncs.h" #include "optimizer/clauses.h" @@ -52,6 +56,9 @@ extern "C" { #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/partcache.h" + +extern bool enable_parallel; +extern int max_parallel_workers_per_gather; } #define GP_WRAP_START \ sigjmp_buf local_sigjmp_buf; \ @@ -2576,6 +2583,19 @@ gpdb::GetForeignServerId(Oid reloid) return 0; } +int16 +gpdb::GetAppendOnlySegmentFilesCount(Relation rel) +{ + GP_WRAP_START; + { + FormData_pg_appendonly aoFormData; + GetAppendOnlyEntry(rel, &aoFormData); + return aoFormData.segfilecount; + } + GP_WRAP_END; + return -1; +} + // Locks on partition leafs and indexes are held during optimizer (after // parse-analyze stage). ORCA need this function to lock relation. Here // we do not need to consider lock-upgrade issue, reasons are: @@ -2734,4 +2754,36 @@ gpdb::TestexprIsHashable(Node *testexpr, List *param_ids) return false; } +// check if parallel mode is OK (comprehensive check) +bool +gpdb::IsParallelModeOK(void) +{ + GP_WRAP_START; + { + if (!enable_parallel) + return false; + + if (IS_SINGLENODE()) + return false; + + if (max_parallel_workers_per_gather <= 0) + return false; + + // Check if parallel plans are enabled in current optimizer context + gpopt::COptCtxt *poctxt = gpopt::COptCtxt::PoctxtFromTLS(); + if (nullptr != poctxt) + { + gpopt::COptimizerConfig *optimizer_config = poctxt->GetOptimizerConfig(); + if (nullptr != optimizer_config) + { + if (!optimizer_config->CreateParallelPlan()) + return false; + } + } + return true; + } + GP_WRAP_END; + return false; // default to disabled if no context +} + // EOF diff --git a/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp b/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp index edce6bfd45a..6084abe1c62 100644 --- a/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp +++ b/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp @@ -30,6 +30,7 @@ extern "C" { #include "partitioning/partdesc.h" #include "storage/lmgr.h" #include "utils/guc.h" +#include "optimizer/cost.h" #include "utils/lsyscache.h" #include "utils/partcache.h" #include "utils/rel.h" @@ -83,6 +84,7 @@ extern "C" { #include "naucrates/dxl/operators/CDXLPhysicalSplit.h" #include "naucrates/dxl/operators/CDXLPhysicalTVF.h" #include "naucrates/dxl/operators/CDXLPhysicalTableScan.h" +#include "naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h" #include "naucrates/dxl/operators/CDXLPhysicalValuesScan.h" #include "naucrates/dxl/operators/CDXLPhysicalWindow.h" #include "naucrates/dxl/operators/CDXLScalarBitmapBoolOp.h" @@ -348,6 +350,12 @@ CTranslatorDXLToPlStmt::TranslateDXLOperatorToPlan( ctxt_translation_prev_siblings); break; } + case EdxlopPhysicalParallelTableScan: + { + plan = TranslateDXLParallelTblScan(dxlnode, output_context, + ctxt_translation_prev_siblings); + break; + } case EdxlopPhysicalIndexScan: { plan = TranslateDXLIndexScan(dxlnode, output_context, @@ -712,6 +720,111 @@ CTranslatorDXLToPlStmt::TranslateDXLTblScan( } +//--------------------------------------------------------------------------- +// @function: +// CTranslatorDXLToPlStmt::TranslateDXLParallelTblScan +// +// @doc: +// Translates a DXL parallel table scan node into a parallel SeqScan node +Plan * +CTranslatorDXLToPlStmt::TranslateDXLParallelTblScan( + const CDXLNode *tbl_scan_dxlnode, CDXLTranslateContext *output_context, + CDXLTranslationContextArray * /*ctxt_translation_prev_siblings*/) +{ + // translate table descriptor into a range table entry + CDXLPhysicalParallelTableScan *phy_parallel_tbl_scan_dxlop = + CDXLPhysicalParallelTableScan::Cast(tbl_scan_dxlnode->GetOperator()); + + ULONG parallel_workers = phy_parallel_tbl_scan_dxlop->UlParallelWorkers(); + + // translation context for column mappings in the base relation + CDXLTranslateContextBaseTable base_table_context(m_mp); + + const CDXLTableDescr *dxl_table_descr = + phy_parallel_tbl_scan_dxlop->GetDXLTableDescr(); + const IMDRelation *md_rel = + m_md_accessor->RetrieveRel(dxl_table_descr->MDId()); + + // Lock any table we are to scan, since it may not have been properly locked + // by the parser (e.g in case of generated scans for partitioned tables) + OID oidRel = CMDIdGPDB::CastMdid(md_rel->MDId())->Oid(); + GPOS_ASSERT(dxl_table_descr->LockMode() != -1); + gpdb::GPDBLockRelationOid(oidRel, dxl_table_descr->LockMode()); + + Index index = ProcessDXLTblDescr(dxl_table_descr, &base_table_context); + + // a table scan node must have 2 children: projection list and filter + GPOS_ASSERT(2 == tbl_scan_dxlnode->Arity()); + + // translate proj list and filter + CDXLNode *project_list_dxlnode = (*tbl_scan_dxlnode)[EdxltsIndexProjList]; + CDXLNode *filter_dxlnode = (*tbl_scan_dxlnode)[EdxltsIndexFilter]; + + List *targetlist = NIL; + + // List to hold the quals after translating filter_dxlnode node. + List *query_quals = NIL; + + TranslateProjListAndFilter( + project_list_dxlnode, filter_dxlnode, + &base_table_context, // translate context for the base table + nullptr, // translate_ctxt_left and pdxltrctxRight, + &targetlist, &query_quals, output_context); + + Plan *plan = nullptr; + Plan *plan_return = nullptr; + + // Parallel table scans are always sequential scans (not foreign scans) + SeqScan *seq_scan = MakeNode(SeqScan); + seq_scan->scanrelid = index; + plan = &(seq_scan->plan); + plan_return = (Plan *) seq_scan; + + // Set parallel execution flags + plan->parallel_aware = true; + plan->parallel_safe = true; + plan->parallel = (int) parallel_workers; + + plan->targetlist = targetlist; + + // List to hold the quals which contain both security quals and query + // quals. + List *security_query_quals = NIL; + + // Fetching the RTE of the relation from the rewritten parse tree + // based on the oidRel and adding the security quals of the RTE in + // the security_query_quals list. + AddSecurityQuals(oidRel, &security_query_quals, &index); + + // The security quals should always be executed first when + // compared to other quals. So appending query quals to the + // security_query_quals list after the security quals. + security_query_quals = + gpdb::ListConcat(security_query_quals, query_quals); + plan->qual = security_query_quals; + + if (md_rel->IsNonBlockTable()) + { + CheckSafeTargetListForAOTables(plan->targetlist); + } + + plan->plan_node_id = m_dxl_to_plstmt_context->GetNextPlanId(); + + // translate operator costs + TranslatePlanCosts(tbl_scan_dxlnode, plan); + + // Adjust row count to per-worker statistics + if (parallel_workers > 1) + { + plan->plan_rows = ceil(plan->plan_rows / parallel_workers); + } + + SetParamIds(plan); + + return plan_return; +} + + //--------------------------------------------------------------------------- // @function: // CTranslatorDXLToPlStmt::SetIndexVarAttnoWalker @@ -719,7 +832,6 @@ CTranslatorDXLToPlStmt::TranslateDXLTblScan( // @doc: // Walker to set index var attno's, // attnos of index vars are set to their relative positions in index keys, -// skip any outer references while walking the expression tree // //--------------------------------------------------------------------------- BOOL @@ -2415,15 +2527,34 @@ CTranslatorDXLToPlStmt::TranslateDXLMotion( sendslice->directDispatch.contentIds = NIL; sendslice->directDispatch.haveProcessedAnyCalculations = false; + // set parallel workers if needed + ULONG child_index = motion_dxlop->GetRelationChildIdx(); + CDXLNode *child_dxlnode = (*motion_dxlnode)[child_index]; + ULONG child_parallel_workers = ExtractParallelWorkersFromDXL(child_dxlnode); + if (child_parallel_workers > 1) + { + // Determine parallel workers based on enable_parallel and gang type + bool supports_parallel = (sendslice->gangType == GANGTYPE_PRIMARY_READER || + sendslice->gangType == GANGTYPE_PRIMARY_WRITER); + + if (supports_parallel) + { + sendslice->parallel_workers = child_parallel_workers; + } + else + { + // Disable parallel for: non-PRIMARY gang types + // (SINGLETON_READER, ENTRYDB_READER, UNALLOCATED) + sendslice->parallel_workers = 0; + } + } + motion->motionID = sendslice->sliceIndex; // translate motion child // child node is in the same position in broadcast and gather motion nodes // but different in redistribute motion nodes - - ULONG child_index = motion_dxlop->GetRelationChildIdx(); - - CDXLNode *child_dxlnode = (*motion_dxlnode)[child_index]; + // Note: child_index and child_dxlnode already defined above CDXLTranslateContext child_context(m_mp, false, output_context->GetColIdToParamIdMap()); @@ -2576,6 +2707,16 @@ CTranslatorDXLToPlStmt::TranslateDXLMotion( return nullptr; } + // Adjust row count for parallel execution in the sending slice + // The Motion node receives rows from all parallel workers, so we need to + // account for the fact that each worker processes a fraction of the rows. + // TranslatePlanCosts() already divided by numsegments, but if we have + // parallel workers, each segment is further subdivided among workers. + if (sendslice->parallel_workers > 1) + { + plan->plan_rows = ceil(plan->plan_rows / sendslice->parallel_workers); + } + SetParamIds(plan); return (Plan *) motion; @@ -7282,4 +7423,75 @@ CTranslatorDXLToPlStmt::IsIndexForOrderBy( } return false; } + +//--------------------------------------------------------------------------- +// @function: +// CTranslatorDXLToPlStmt::ExtractParallelWorkersFromDXL +// +// @doc: +// Extract parallel workers count from DXL node tree recursively. +// Since parallel degree is uniform across all parallel scans in a query, +// returns the first parallel degree found from any CDXLPhysicalParallelTableScan, +// or 1 if no parallel scan exists. +// +//--------------------------------------------------------------------------- +ULONG +CTranslatorDXLToPlStmt::ExtractParallelWorkersFromDXL(const CDXLNode *dxlnode) +{ + if (nullptr == dxlnode) + { + return 1; + } + + CDXLOperator *dxlop = dxlnode->GetOperator(); + if (EdxlopPhysicalParallelTableScan == dxlop->GetDXLOperator()) + { + // Return parallel workers from the parallel table scan operator + // All parallel scans in the query share the same parallel degree + CDXLPhysicalParallelTableScan *parallel_scan_dxlop = + CDXLPhysicalParallelTableScan::Cast(dxlop); + return parallel_scan_dxlop->UlParallelWorkers(); + } + else if (EdxlopPhysicalTableScan == dxlop->GetDXLOperator() || + EdxlopPhysicalDynamicTableScan == dxlop->GetDXLOperator() || + EdxlopPhysicalIndexScan == dxlop->GetDXLOperator() || + EdxlopPhysicalIndexOnlyScan == dxlop->GetDXLOperator() || + EdxlopPhysicalBitmapTableScan == dxlop->GetDXLOperator() || + EdxlopPhysicalDynamicBitmapTableScan == dxlop->GetDXLOperator() || + EdxlopPhysicalForeignScan == dxlop->GetDXLOperator() || + EdxlopPhysicalDynamicForeignScan == dxlop->GetDXLOperator() || + EdxlopPhysicalDynamicIndexScan == dxlop->GetDXLOperator() || + EdxlopPhysicalDynamicIndexOnlyScan == dxlop->GetDXLOperator() || + EdxlopPhysicalValuesScan == dxlop->GetDXLOperator()) + { + // Non-parallel scans (table, index, bitmap, foreign, values) + // These are leaf nodes in terms of parallel worker extraction + // Return 1 to indicate no parallel workers + return 1; + } + else if (EdxlopPhysicalMotionGather == dxlop->GetDXLOperator() || + EdxlopPhysicalMotionBroadcast == dxlop->GetDXLOperator() || + EdxlopPhysicalMotionRedistribute == dxlop->GetDXLOperator() || + EdxlopPhysicalMotionRandom == dxlop->GetDXLOperator() || + EdxlopPhysicalMotionRoutedDistribute == dxlop->GetDXLOperator()) + { + // Motion node creates a slice boundary - do not recurse into child + // The child's parallel workers belong to the sending slice, not receiving slice + // Return 0 to indicate the receiving slice (current slice) has no parallel workers + return 1; + } + + // Recursively check child nodes, return early when first parallel scan is found + for (ULONG ul = 0; ul < dxlnode->Arity(); ul++) + { + ULONG child_parallel_workers = ExtractParallelWorkersFromDXL((*dxlnode)[ul]); + if (child_parallel_workers > 1) + { + return child_parallel_workers; + } + } + + return 1; +} + // EOF diff --git a/src/backend/gpopt/translate/CTranslatorRelcacheToDXL.cpp b/src/backend/gpopt/translate/CTranslatorRelcacheToDXL.cpp index ee4b8888b19..aa23d3932a2 100644 --- a/src/backend/gpopt/translate/CTranslatorRelcacheToDXL.cpp +++ b/src/backend/gpopt/translate/CTranslatorRelcacheToDXL.cpp @@ -516,7 +516,6 @@ CTranslatorRelcacheToDXL::RetrieveRel(CMemoryPool *mp, CMDAccessor *md_accessor, IMdIdArray *check_constraint_mdids = nullptr; BOOL is_temporary = false; BOOL is_partitioned = false; - IMDRelation *md_rel = nullptr; IMdIdArray *partition_oids = nullptr; IMDId *foreign_server_mdid = nullptr; @@ -618,14 +617,31 @@ CTranslatorRelcacheToDXL::RetrieveRel(CMemoryPool *mp, CMDAccessor *md_accessor, CMDIdGPDB(IMDId::EmdidGeneral, gpdb::GetForeignServerId(oid)); } - md_rel = GPOS_NEW(mp) CMDRelationGPDB( + CMDRelationGPDB *md_rel_gpdb = GPOS_NEW(mp) CMDRelationGPDB( mp, mdid, mdname, is_temporary, rel_storage_type, dist, mdcol_array, distr_cols, distr_op_families, part_keys, part_types, partition_oids, convert_hash_to_random, keyset_array, md_index_info_array, check_constraint_mdids, mdpart_constraint, foreign_server_mdid, rel->rd_rel->reltuples); - return md_rel; + // Set segment file count for AO/AOCO tables + // Skip partitioned tables as they don't have physical storage (only leaf partitions do) + if ((rel_storage_type == IMDRelation::ErelstorageAppendOnlyRows || + rel_storage_type == IMDRelation::ErelstorageAppendOnlyCols) && + rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) + { + INT seg_file_count = gpdb::GetAppendOnlySegmentFilesCount(rel.get()); + md_rel_gpdb->SetSegFileCount(seg_file_count); + } + + // Set parallel workers from table options + if (rel->rd_options != NULL) + { + INT parallel_workers = RelationGetParallelWorkers(rel.get(), -1); + md_rel_gpdb->SetParallelWorkers(parallel_workers); + } + + return md_rel_gpdb; } //--------------------------------------------------------------------------- diff --git a/src/backend/gpopt/utils/COptTasks.cpp b/src/backend/gpopt/utils/COptTasks.cpp index 6bd93379f0f..30e6e131f3d 100644 --- a/src/backend/gpopt/utils/COptTasks.cpp +++ b/src/backend/gpopt/utils/COptTasks.cpp @@ -385,7 +385,7 @@ COptTasks::LoadSearchStrategy(CMemoryPool *mp, char *path) //--------------------------------------------------------------------------- COptimizerConfig * COptTasks::CreateOptimizerConfig(CMemoryPool *mp, ICostModel *cost_model, - CPlanHint *plan_hints) + CPlanHint *plan_hints, BOOL enable_parallel_plans) { // get chosen plan number, cost threshold ULLONG plan_id = (ULLONG) optimizer_plan_id; @@ -424,7 +424,8 @@ COptTasks::CreateOptimizerConfig(CMemoryPool *mp, ICostModel *cost_model, push_group_by_below_setop_threshold, xform_bind_threshold, skew_factor), plan_hints, - GPOS_NEW(mp) CWindowOids(mp, OID(F_ROW_NUMBER), OID(F_RANK_), OID(F_DENSE_RANK_))); + GPOS_NEW(mp) CWindowOids(mp, OID(F_ROW_NUMBER), OID(F_RANK_), OID(F_DENSE_RANK_)), + enable_parallel_plans); } //--------------------------------------------------------------------------- @@ -961,7 +962,7 @@ COptTasks::OptimizeTask(void *ptr) ICostModel *cost_model = GetCostModel(mp, num_segments_for_costing); CPlanHint *plan_hints = GetPlanHints(mp, opt_ctxt->m_query); COptimizerConfig *optimizer_config = - CreateOptimizerConfig(mp, cost_model, plan_hints); + CreateOptimizerConfig(mp, cost_model, plan_hints, opt_ctxt->m_create_parallel_plan); CConstExprEvaluatorProxy expr_eval_proxy(mp, &mda); IConstExprEvaluator *expr_evaluator = GPOS_NEW(mp) CConstExprEvaluatorDXL(mp, &mda, &expr_eval_proxy); @@ -1186,6 +1187,7 @@ COptTasks::GPOPTOptimizedPlan(Query *query, SOptContext *gpopt_context, Optimize gpopt_context->m_should_generate_plan_stmt = true; // Copy options in `OptimizerOptions` to `SOptContext` gpopt_context->m_create_vec_plan = opts->create_vectorization_plan; + gpopt_context->m_create_parallel_plan = opts->create_parallel_plan; Execute(&OptimizeTask, gpopt_context); return gpopt_context->m_plan_stmt; } diff --git a/src/backend/gporca/libgpdbcost/include/gpdbcost/CCostModelGPDB.h b/src/backend/gporca/libgpdbcost/include/gpdbcost/CCostModelGPDB.h index 2b44693fa4e..7db2cdbbdd1 100644 --- a/src/backend/gporca/libgpdbcost/include/gpdbcost/CCostModelGPDB.h +++ b/src/backend/gporca/libgpdbcost/include/gpdbcost/CCostModelGPDB.h @@ -80,6 +80,11 @@ private: const CCostModelGPDB *pcmgpdb, const SCostingInfo *pci); + // cost of parallel table scan + static CCost CostParallelTableScan(CMemoryPool *mp, CExpressionHandle &exprhdl, + const CCostModelGPDB *pcmgpdb, + const SCostingInfo *pci); + // cost of filter static CCost CostFilter(CMemoryPool *mp, CExpressionHandle &exprhdl, const CCostModelGPDB *pcmgpdb, @@ -225,6 +230,10 @@ private: IStatistics *&stats, CMDAccessor *md_accessor, CMemoryPool *mp); + // Helper functions for parallel cost calculation + static CDouble CalculateParallelEfficiency(ULONG ulWorkers); + static CDouble GetWorkerStartupCost(const CCostModelGPDB *pcmgpdb, ULONG ulWorkers); + public: // ctor CCostModelGPDB(CMemoryPool *mp, ULONG ulSegments, diff --git a/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp b/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp index 0bf5e167469..575f6473ff0 100644 --- a/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp +++ b/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp @@ -12,6 +12,7 @@ #include "gpdbcost/CCostModelGPDB.h" #include <limits> +#include <cmath> #include "gpopt/base/CColRefSetIter.h" #include "gpopt/base/COptCtxt.h" @@ -27,6 +28,7 @@ #include "gpopt/operators/CPhysicalHashAgg.h" #include "gpopt/operators/CPhysicalIndexOnlyScan.h" #include "gpopt/operators/CPhysicalIndexScan.h" +#include "gpopt/operators/CPhysicalParallelTableScan.h" #include "gpopt/operators/CPhysicalMotion.h" #include "gpopt/operators/CPhysicalMotionBroadcast.h" #include "gpopt/operators/CPhysicalPartitionSelector.h" @@ -43,6 +45,8 @@ using namespace gpos; using namespace gpdbcost; +// Forward declare PostgreSQL GUC variables +extern double parallel_setup_cost; //--------------------------------------------------------------------------- // @function: @@ -2380,7 +2384,8 @@ CCostModelGPDB::CostScan(CMemoryPool *, // mp GPOS_ASSERT(COperator::EopPhysicalTableScan == op_id || COperator::EopPhysicalDynamicTableScan == op_id || COperator::EopPhysicalForeignScan == op_id || - COperator::EopPhysicalDynamicForeignScan == op_id); + COperator::EopPhysicalDynamicForeignScan == op_id || + COperator::EopPhysicalParallelTableScan == op_id); const CDouble dInitScan = pcmgpdb->GetCostModelParams() @@ -2402,6 +2407,7 @@ CCostModelGPDB::CostScan(CMemoryPool *, // mp case COperator::EopPhysicalDynamicTableScan: case COperator::EopPhysicalForeignScan: case COperator::EopPhysicalDynamicForeignScan: + case COperator::EopPhysicalParallelTableScan: // table scan cost considers only retrieving tuple cost, // since we scan the entire table here, the cost is correlated with table rows and table width, // since Scan's parent operator may be a filter that will be pushed into Scan node in GPDB plan, @@ -2416,6 +2422,120 @@ CCostModelGPDB::CostScan(CMemoryPool *, // mp } +//--------------------------------------------------------------------------- +// @function: +// CCostModelGPDB::CostParallelTableScan +// +// @doc: +// Cost of parallel table scan +// +//--------------------------------------------------------------------------- +CCost +CCostModelGPDB::CostParallelTableScan(CMemoryPool *mp, + CExpressionHandle &exprhdl, + const CCostModelGPDB *pcmgpdb, + const SCostingInfo *pci) +{ + GPOS_ASSERT(nullptr != pcmgpdb); + GPOS_ASSERT(nullptr != pci); + + COperator *pop = exprhdl.Pop(); + GPOS_ASSERT(COperator::EopPhysicalParallelTableScan == pop->Eopid()); + + // Get the parallel table scan operator + CPhysicalParallelTableScan *popParallelScan = + CPhysicalParallelTableScan::PopConvert(pop); + ULONG ulWorkers = popParallelScan->UlParallelWorkers(); + + // If only 1 worker, use regular scan cost + if (ulWorkers <= 1) + { + return CostScan(mp, exprhdl, pcmgpdb, pci); + } + + // Get base scan parameters + const CDouble dInitScan = + pcmgpdb->GetCostModelParams() + ->PcpLookup(CCostModelParamsGPDB::EcpInitScanFactor) + ->Get(); + const CDouble dTableWidth = + CPhysicalScan::PopConvert(pop)->PstatsBaseTable()->Width(); + const CDouble dTableScanCostUnit = + pcmgpdb->GetCostModelParams() + ->PcpLookup(CCostModelParamsGPDB::EcpTableScanCostUnit) + ->Get(); + + // Calculate base scan cost + CDouble dBaseScanCost = dInitScan + pci->Rows() * dTableWidth * dTableScanCostUnit; + + // Calculate parallel efficiency (decreases with more workers) + CDouble dParallelEfficiency = CalculateParallelEfficiency(ulWorkers); + + // Parallel scan cost = base cost / (workers * efficiency) + CDouble dParallelScanCost = dBaseScanCost / (ulWorkers * dParallelEfficiency); + + // Add worker startup cost + CDouble dWorkerStartupCost = GetWorkerStartupCost(pcmgpdb, ulWorkers); + + // Total cost + return CCost(pci->NumRebinds() * dParallelScanCost + dWorkerStartupCost); +} + +//--------------------------------------------------------------------------- +// @function: +// CCostModelGPDB::CalculateParallelEfficiency +// +// @doc: +// Calculate parallel efficiency factor (0-1) based on worker count +// +//--------------------------------------------------------------------------- +CDouble +CCostModelGPDB::CalculateParallelEfficiency(ULONG ulWorkers) +{ + if (ulWorkers <= 1) + { + return 1.0; + } + + // Efficiency decreases logarithmically with more workers + // Formula: efficiency = 1 / (1 + 0.1 * log2(workers)) + // This gives: 2 workers = 0.91, 4 workers = 0.83, 8 workers = 0.77 + double dLogWorkers = std::log2(static_cast<double>(ulWorkers)); + return CDouble(1.0 / (1.0 + 0.1 * dLogWorkers)); +} + + +//--------------------------------------------------------------------------- +// @function: +// CCostModelGPDB::GetWorkerStartupCost +// +// @doc: +// Get the cost of starting up parallel workers +// +//--------------------------------------------------------------------------- +CDouble +CCostModelGPDB::GetWorkerStartupCost(const CCostModelGPDB * /* pcmgpdb */, ULONG ulWorkers) +{ + if (ulWorkers <= 1) + { + return 0.0; + } + + // ORCA's cost units are much smaller than PostgreSQL's cost model + // PostgreSQL's parallel_setup_cost default is 1000, but ORCA's costs are: + // - InitScanFactor: 431.0 + // - HJHashTableInitCostFactor: 500.0 + // - DefaultCost: 100.0 + // + // Use a conversion factor to map parallel_setup_cost to ORCA's scale. + // With default parallel_setup_cost=1000, this gives 10.0, which is + // reasonable compared to InitScanFactor (431.0) - about 0.1% overhead + const double POSTGRES_TO_ORCA_COST_CONVERSION = 0.001; + + return CDouble(parallel_setup_cost * POSTGRES_TO_ORCA_COST_CONVERSION); +} + + //--------------------------------------------------------------------------- // @function: // CCostModelGPDB::CostFilter @@ -2489,11 +2609,15 @@ CCostModelGPDB::Cost( case COperator::EopPhysicalDynamicTableScan: case COperator::EopPhysicalForeignScan: case COperator::EopPhysicalDynamicForeignScan: - { return CostScan(m_mp, exprhdl, this, pci); } + case COperator::EopPhysicalParallelTableScan: + { + return CostParallelTableScan(m_mp, exprhdl, this, pci); + } + case COperator::EopPhysicalFilter: { return CostFilter(m_mp, exprhdl, this, pci); diff --git a/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpec.h b/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpec.h index d3b47b95f80..9f3b74bd76f 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpec.h +++ b/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpec.h @@ -71,6 +71,7 @@ public: EdtStrictSingleton, // data is on a single segment or the master (derived only, only compatible with other singleton distributions) EdtRandom, // data is randomly distributed across all segments EdtStrictRandom, // same as random, used to force multiple slices for parallel union all. + EdtWorkerRandom, // data is randomly distributed among parallel workers within segments EdtRouted, // data is routed to a segment explicitly specified in the tuple, EdtUniversal, // data is available everywhere (derived only) EdtNonSingleton, // data can have any distribution except singleton (required only) diff --git a/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpecWorkerRandom.h b/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpecWorkerRandom.h new file mode 100644 index 00000000000..f026f00e247 --- /dev/null +++ b/src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpecWorkerRandom.h @@ -0,0 +1,139 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CDistributionSpecWorkerRandom.h + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/include/gpopt/base/CDistributionSpecWorkerRandom.h + * + *------------------------------------------------------------------------- + */ +#ifndef GPOPT_CDistributionSpecWorkerRandom_H +#define GPOPT_CDistributionSpecWorkerRandom_H + +#include "gpos/base.h" + +#include "gpopt/base/CDistributionSpecRandom.h" + +namespace gpopt +{ +using namespace gpos; + +//--------------------------------------------------------------------------- +// @class: +// CDistributionSpecWorkerRandom +// +// @doc: +// Class for representing worker-level random distribution. +// This class provides a specialized implementation for parallel +// worker execution with explicit worker count management. +// +//--------------------------------------------------------------------------- +class CDistributionSpecWorkerRandom : public CDistributionSpecRandom +{ +private: + // Number of workers for parallel execution + ULONG m_ulWorkers; + + // Base segment distribution (usually segment-level random) + CDistributionSpec *m_pdsSegmentBase; + + // private copy ctor + CDistributionSpecWorkerRandom(const CDistributionSpecWorkerRandom &); + +public: + // ctor + CDistributionSpecWorkerRandom(ULONG ulWorkers, CDistributionSpec *pdsSegmentBase = nullptr); + + // dtor + ~CDistributionSpecWorkerRandom() override; + + // distribution type accessor + EDistributionType + Edt() const override + { + return CDistributionSpec::EdtWorkerRandom; + } + + // distribution identifier + const CHAR * + SzId() const override + { + return "WORKER_RANDOM"; + } + + // Get worker count + ULONG + UlWorkers() const + { + return m_ulWorkers; + } + + // Get base segment distribution + CDistributionSpec * + PdsSegmentBase() const + { + return m_pdsSegmentBase; + } + + // does this distribution match the given one + BOOL Matches(const CDistributionSpec *pds) const override; + + // does this distribution satisfy the given one + BOOL FSatisfies(const CDistributionSpec *pds) const override; + + // append enforcers to dynamic array for the given plan properties + void AppendEnforcers(CMemoryPool *mp, CExpressionHandle &exprhdl, + CReqdPropPlan *prpp, CExpressionArray *pdrgpexpr, + CExpression *pexpr) override; + + // print + IOstream &OsPrint(IOstream &os) const override; + + // Factory method for creating worker-level random distribution + static CDistributionSpecWorkerRandom *PdsCreateWorkerRandom( + CMemoryPool *mp, ULONG ulWorkers, CDistributionSpec *pdsBase = nullptr); + + // conversion function + static CDistributionSpecWorkerRandom * + PdsConvert(CDistributionSpec *pds) + { + GPOS_ASSERT(nullptr != pds); + GPOS_ASSERT(EdtWorkerRandom == pds->Edt()); + + return dynamic_cast<CDistributionSpecWorkerRandom *>(pds); + } + + // conversion function: const argument + static const CDistributionSpecWorkerRandom * + PdsConvert(const CDistributionSpec *pds) + { + GPOS_ASSERT(nullptr != pds); + GPOS_ASSERT(EdtWorkerRandom == pds->Edt()); + + return dynamic_cast<const CDistributionSpecWorkerRandom *>(pds); + } + +}; // class CDistributionSpecWorkerRandom + +} // namespace gpopt + +#endif // !GPOPT_CDistributionSpecWorkerRandom_H + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/include/gpopt/base/CRewindabilitySpec.h b/src/backend/gporca/libgpopt/include/gpopt/base/CRewindabilitySpec.h index bd2dea7eec1..2959a678301 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/base/CRewindabilitySpec.h +++ b/src/backend/gporca/libgpopt/include/gpopt/base/CRewindabilitySpec.h @@ -108,6 +108,7 @@ public: EmhtSentinel }; + private: // rewindability support ERewindabilityType m_rewindability; diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h b/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h index 657e3082ef7..5d10a1cee99 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h +++ b/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h @@ -188,6 +188,7 @@ public: EopScalarFieldSelect, EopPhysicalTableScan, + EopPhysicalParallelTableScan, EopPhysicalForeignScan, EopPhysicalIndexScan, EopPhysicalIndexOnlyScan, diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelTableScan.h new file mode 100644 index 00000000000..2ebb916d4f3 --- /dev/null +++ b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelTableScan.h @@ -0,0 +1,140 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CPhysicalParallelTableScan.h + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelTableScan.h + * + *------------------------------------------------------------------------- + */ +#ifndef GPOPT_CPhysicalParallelTableScan_H +#define GPOPT_CPhysicalParallelTableScan_H + +#include "gpos/base.h" + +#include "gpopt/operators/CPhysicalTableScan.h" + +namespace gpopt +{ +//--------------------------------------------------------------------------- +// @class: +// CPhysicalParallelTableScan +// +// @doc: +// Parallel table scan operator +// +//--------------------------------------------------------------------------- +class CPhysicalParallelTableScan : public CPhysicalTableScan +{ +private: + // number of parallel workers + ULONG m_ulParallelWorkers; + + // worker-level distribution spec + CDistributionSpec *m_pdsWorkerDistribution; + + // private copy ctor + CPhysicalParallelTableScan(const CPhysicalParallelTableScan &); + +public: + // ctors + explicit CPhysicalParallelTableScan(CMemoryPool *mp); + CPhysicalParallelTableScan(CMemoryPool *mp, const CName *pnameAlias, + CTableDescriptor *ptabdesc, + CColRefArray *pdrgpcrOutput, + ULONG ulParallelWorkers); + + // dtor + ~CPhysicalParallelTableScan() override; + + // ident accessors + EOperatorId + Eopid() const override + { + return EopPhysicalParallelTableScan; + } + + // return a string for operator name + const CHAR * + SzId() const override + { + return "CPhysicalParallelTableScan"; + } + + // number of parallel workers + ULONG UlParallelWorkers() const + { + return m_ulParallelWorkers; + } + + // operator specific hash function + ULONG HashValue() const override; + + // match function + BOOL Matches(COperator *) const override; + + // debug print + IOstream &OsPrint(IOstream &) const override; + + // conversion function + static CPhysicalParallelTableScan * + PopConvert(COperator *pop) + { + GPOS_ASSERT(nullptr != pop); + GPOS_ASSERT(EopPhysicalParallelTableScan == pop->Eopid()); + + return dynamic_cast<CPhysicalParallelTableScan *>(pop); + } + + CRewindabilitySpec * + PrsDerive(CMemoryPool *mp, + CExpressionHandle & // exprhdl + ) const override + { + return GPOS_NEW(mp) + CRewindabilitySpec(CRewindabilitySpec::ErtNone, + CRewindabilitySpec::EmhtNoMotion); + } + + // derive distribution + CDistributionSpec *PdsDerive(CMemoryPool *mp, CExpressionHandle &exprhdl) const override; + + // return distribution property enforcing type for this operator + CEnfdProp::EPropEnforcingType EpetDistribution( + CExpressionHandle &exprhdl, + const CEnfdDistribution *ped) const override; + + // return rewindability property enforcing type for this operator + CEnfdProp::EPropEnforcingType EpetRewindability( + CExpressionHandle &exprhdl, + const CEnfdRewindability *per) const override; + + // check if optimization contexts is valid + // Reject if parent requires REWINDABLE (e.g., for NL Join inner child) + BOOL FValidContext(CMemoryPool *mp, COptimizationContext *poc, + COptimizationContextArray *pdrgpocChild) const override; + +}; // class CPhysicalParallelTableScan + +} // namespace gpopt + +#endif // !GPOPT_CPhysicalParallelTableScan_H + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalTableScan.h index ace1005d67a..d34f0150b55 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalTableScan.h +++ b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalTableScan.h @@ -71,7 +71,8 @@ public: { GPOS_ASSERT(nullptr != pop); GPOS_ASSERT(EopPhysicalTableScan == pop->Eopid() || - EopPhysicalForeignScan == pop->Eopid()); + EopPhysicalForeignScan == pop->Eopid() || + EopPhysicalParallelTableScan == pop->Eopid()); return dynamic_cast<CPhysicalTableScan *>(pop); } diff --git a/src/backend/gporca/libgpopt/include/gpopt/optimizer/COptimizerConfig.h b/src/backend/gporca/libgpopt/include/gpopt/optimizer/COptimizerConfig.h index 159ddafbed7..eeee3932b3e 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/optimizer/COptimizerConfig.h +++ b/src/backend/gporca/libgpopt/include/gpopt/optimizer/COptimizerConfig.h @@ -69,11 +69,15 @@ private: // default window oids CWindowOids *m_window_oids; + // should generate parallel plans ? + BOOL m_create_parallel_plan; + public: // ctor COptimizerConfig(CEnumeratorConfig *pec, CStatisticsConfig *stats_config, CCTEConfig *pcteconf, ICostModel *pcm, CHint *phint, - CPlanHint *pplanhint, CWindowOids *pdefoidsGPDB); + CPlanHint *pplanhint, CWindowOids *pdefoidsGPDB, + BOOL enable_parallel_plans = false); // dtor ~COptimizerConfig() override; @@ -127,6 +131,13 @@ public: return m_plan_hint; } + // parallel plans setting + BOOL + CreateParallelPlan() const + { + return m_create_parallel_plan; + } + // generate default optimizer configurations static COptimizerConfig *PoconfDefault(CMemoryPool *mp); diff --git a/src/backend/gporca/libgpopt/include/gpopt/search/CGroup.h b/src/backend/gporca/libgpopt/include/gpopt/search/CGroup.h index 0b94e9df7e7..eaeaf22024f 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/search/CGroup.h +++ b/src/backend/gporca/libgpopt/include/gpopt/search/CGroup.h @@ -38,6 +38,7 @@ class CDrvdProp; class CDrvdPropCtxtPlan; class CReqdPropRelational; class CExpression; +class CMemo; // type definitions // array of groups @@ -160,6 +161,9 @@ private: // memory pool CMemoryPool *m_mp; + // containing memo + CMemo *m_pmemo; + // id is used when printing memo contents ULONG m_id; @@ -257,6 +261,9 @@ private: // setter of group state void SetState(EState estNewState); + // setter of containing memo + void SetMemo(CMemo *pmemo); + // set hash join keys void SetJoinKeys(CExpressionArray *pdrgpexprOuter, CExpressionArray *pdrgpexprInner, @@ -338,6 +345,13 @@ public: return m_id; } + // containing memo accessor + CMemo * + Pmemo() const + { + return m_pmemo; + } + // group properties accessor CDrvdProp * Pdp() const diff --git a/src/backend/gporca/libgpopt/include/gpopt/search/CGroupProxy.h b/src/backend/gporca/libgpopt/include/gpopt/search/CGroupProxy.h index 61303f39ef7..1e77f0fbcab 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/search/CGroupProxy.h +++ b/src/backend/gporca/libgpopt/include/gpopt/search/CGroupProxy.h @@ -23,6 +23,7 @@ using namespace gpos; class CGroupExpression; class CDrvdProp; class COptimizationContext; +class CMemo; //--------------------------------------------------------------------------- // @class: @@ -63,6 +64,13 @@ public: m_pgroup->SetState(estNewState); } + // set containing memo + void + SetMemo(CMemo *pmemo) + { + m_pgroup->SetMemo(pmemo); + } + // set hash join keys void SetJoinKeys(CExpressionArray *pdrgpexprOuter, diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h index 225371097c5..d0fca8219ac 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h +++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h @@ -69,6 +69,7 @@ public: ExfExpandNAryJoinMinCard, ExfExpandNAryJoinDP, ExfGet2TableScan, + ExfGet2ParallelTableScan, ExfIndexGet2IndexScan, ExfDynamicGet2DynamicTableScan, ExfDynamicIndexGet2DynamicIndexScan, diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h new file mode 100644 index 00000000000..99c8d4863d4 --- /dev/null +++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h @@ -0,0 +1,88 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CXformGet2ParallelTableScan.h + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h + * + *------------------------------------------------------------------------- + */ +#ifndef GPOPT_CXformGet2ParallelTableScan_H +#define GPOPT_CXformGet2ParallelTableScan_H + +#include "gpos/base.h" + +#include "gpopt/xforms/CXformImplementation.h" + +namespace gpopt +{ +using namespace gpos; + +//--------------------------------------------------------------------------- +// @class: +// CXformGet2ParallelTableScan +// +// @doc: +// Transform Get to Parallel TableScan using GUC enable_parallel +// +//--------------------------------------------------------------------------- +class CXformGet2ParallelTableScan : public CXformImplementation +{ +private: + // check if memo contains logical operators that are incompatible with parallel execution + static BOOL FHasParallelIncompatibleOps(CExpressionHandle &exprhdl); + +public: + CXformGet2ParallelTableScan(const CXformGet2ParallelTableScan &) = delete; + + // ctor + explicit CXformGet2ParallelTableScan(CMemoryPool *); + + // dtor + ~CXformGet2ParallelTableScan() override = default; + + // ident accessors + EXformId + Exfid() const override + { + return ExfGet2ParallelTableScan; + } + + // return a string for xform name + const CHAR * + SzId() const override + { + return "CXformGet2ParallelTableScan"; + } + + // compute xform promise for a given expression handle + EXformPromise Exfp(CExpressionHandle &exprhdl) const override; + + // actual transform + void Transform(CXformContext *pxfctxt, CXformResult *pxfres, + CExpression *pexpr) const override; + +}; // class CXformGet2ParallelTableScan + +} // namespace gpopt + +#endif // !GPOPT_CXformGet2ParallelTableScan_H + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h index fb5ba6b1e3c..0ee9beb2eb7 100644 --- a/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h +++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h @@ -52,6 +52,7 @@ #include "gpopt/xforms/CXformGbAggDedup2StreamAggDedup.h" #include "gpopt/xforms/CXformGbAggWithMDQA2Join.h" #include "gpopt/xforms/CXformGet2TableScan.h" +#include "gpopt/xforms/CXformGet2ParallelTableScan.h" #include "gpopt/xforms/CXformImplementAssert.h" #include "gpopt/xforms/CXformImplementBitmapTableGet.h" #include "gpopt/xforms/CXformImplementCTEConsumer.h" diff --git a/src/backend/gporca/libgpopt/src/base/CDistributionSpecWorkerRandom.cpp b/src/backend/gporca/libgpopt/src/base/CDistributionSpecWorkerRandom.cpp new file mode 100644 index 00000000000..a5e9e86b02d --- /dev/null +++ b/src/backend/gporca/libgpopt/src/base/CDistributionSpecWorkerRandom.cpp @@ -0,0 +1,330 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CDistributionSpecWorkerRandom.cpp + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/src/base/CDistributionSpecWorkerRandom.cpp + * + *------------------------------------------------------------------------- + */ + +#include "gpopt/base/CDistributionSpecWorkerRandom.h" + +#include "gpopt/base/CColRefSet.h" +#include "gpopt/base/CDistributionSpecHashed.h" +#include "gpopt/base/CDistributionSpecStrictRandom.h" +#include "gpopt/base/COptCtxt.h" +#include "gpopt/base/CUtils.h" +#include "gpopt/operators/CExpressionHandle.h" +#include "gpopt/operators/CPhysicalMotionHashDistribute.h" +#include "gpopt/operators/CPhysicalMotionRandom.h" +#include "naucrates/traceflags/traceflags.h" + +using namespace gpopt; + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::CDistributionSpecWorkerRandom +// +// @doc: +// Ctor +// Note: This constructor should only be called from PdsCreateWorkerRandom +// factory method, which ensures pdsSegmentBase is properly initialized +// +//--------------------------------------------------------------------------- +CDistributionSpecWorkerRandom::CDistributionSpecWorkerRandom(ULONG ulWorkers, CDistributionSpec *pdsSegmentBase) + : m_ulWorkers(ulWorkers), m_pdsSegmentBase(pdsSegmentBase) +{ + GPOS_ASSERT(ulWorkers > 0); + GPOS_ASSERT(nullptr != pdsSegmentBase && + "pdsSegmentBase must be non-null. Use PdsCreateWorkerRandom factory method."); + + m_pdsSegmentBase->AddRef(); + + if (COptCtxt::PoctxtFromTLS()->FDMLQuery()) + { + // set duplicate sensitive flag to enforce Hash-Distribution of + // Const Tables in DML queries + MarkDuplicateSensitive(); + } +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::~CDistributionSpecWorkerRandom +// +// @doc: +// Dtor +// +//--------------------------------------------------------------------------- +CDistributionSpecWorkerRandom::~CDistributionSpecWorkerRandom() +{ + CRefCount::SafeRelease(m_pdsSegmentBase); +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::PdsCreateWorkerRandom +// +// @doc: +// Factory method for creating worker-level random distribution +// +//--------------------------------------------------------------------------- +CDistributionSpecWorkerRandom * +CDistributionSpecWorkerRandom::PdsCreateWorkerRandom(CMemoryPool *mp, ULONG ulWorkers, CDistributionSpec *pdsBase) +{ + GPOS_ASSERT(nullptr != mp); + GPOS_ASSERT(ulWorkers > 0); + + // If no base distribution provided, create a default random distribution + // using the provided memory pool (not TLS pool) + CDistributionSpec *pdsSegmentBase = pdsBase; + if (nullptr == pdsSegmentBase) + { + pdsSegmentBase = GPOS_NEW(mp) CDistributionSpecRandom(); + } + + return GPOS_NEW(mp) CDistributionSpecWorkerRandom(ulWorkers, pdsSegmentBase); +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::Matches +// +// @doc: +// Match function +// +//--------------------------------------------------------------------------- +BOOL +CDistributionSpecWorkerRandom::Matches(const CDistributionSpec *pds) const +{ + if (pds->Edt() == CDistributionSpec::EdtWorkerRandom) + { + const CDistributionSpecWorkerRandom *pdsWorkerRandom = + CDistributionSpecWorkerRandom::PdsConvert(pds); + + // Check if worker counts match and base distributions are compatible + return (m_ulWorkers == pdsWorkerRandom->m_ulWorkers && + IsDuplicateSensitive() == pdsWorkerRandom->IsDuplicateSensitive() && + ((nullptr == m_pdsSegmentBase && nullptr == pdsWorkerRandom->m_pdsSegmentBase) || + (nullptr != m_pdsSegmentBase && nullptr != pdsWorkerRandom->m_pdsSegmentBase && + m_pdsSegmentBase->Matches(pdsWorkerRandom->m_pdsSegmentBase)))); + } + + return false; +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::FSatisfies +// +// @doc: +// Check if this distribution spec satisfies the given one +// +//--------------------------------------------------------------------------- +BOOL +CDistributionSpecWorkerRandom::FSatisfies(const CDistributionSpec *pds) const +{ + if (Matches(pds)) + { + return true; + } + + // Handle different distribution types + if (EdtWorkerRandom == pds->Edt()) + { + const CDistributionSpecWorkerRandom *pdsWorkerRandom = + CDistributionSpecWorkerRandom::PdsConvert(pds); + + // Worker-level can satisfy another worker-level if it has the same number of workers + // and the base segment distribution is compatible + return (m_ulWorkers == pdsWorkerRandom->m_ulWorkers && + (nullptr == m_pdsSegmentBase || nullptr == pdsWorkerRandom->m_pdsSegmentBase || + m_pdsSegmentBase->FSatisfies(pdsWorkerRandom->m_pdsSegmentBase)) && + (IsDuplicateSensitive() || !pdsWorkerRandom->IsDuplicateSensitive())); + } + else if (EdtRandom == pds->Edt()) + { + // Note: This ensures semantic consistency with Random::FSatisfies(WorkerRandom), + // which also returns false, as neither can satisfy the other without a Motion. + return false; + } + + // Standard satisfaction logic for other distribution types + return EdtAny == pds->Edt() || EdtNonSingleton == pds->Edt() || + EdtNonReplicated == pds->Edt(); +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::AppendEnforcers +// +// @doc: +// Add required enforcers to dynamic array +// +//--------------------------------------------------------------------------- +void +CDistributionSpecWorkerRandom::AppendEnforcers(CMemoryPool *mp, + CExpressionHandle &exprhdl, + CReqdPropPlan *prpp, + CExpressionArray *pdrgpexpr, + CExpression *pexpr) +{ + GPOS_ASSERT(nullptr != mp); + GPOS_ASSERT(nullptr != prpp); + GPOS_ASSERT(nullptr != pdrgpexpr); + GPOS_ASSERT(nullptr != pexpr); + GPOS_ASSERT(!GPOS_FTRACE(EopttraceDisableMotions)); + GPOS_ASSERT( + this == prpp->Ped()->PdsRequired() && + "required plan properties don't match enforced distribution spec"); + + // Get the actually required distribution specification + CDistributionSpec *pdsRequired = prpp->Ped()->PdsRequired(); + GPOS_ASSERT(nullptr != pdsRequired); + + // Get child's distribution for duplicate hazard checking + CDistributionSpec *expr_dist_spec = + CDrvdPropPlan::Pdpplan(exprhdl.Pdp())->Pds(); + BOOL fDuplicateHazard = CUtils::FDuplicateHazardDistributionSpec(expr_dist_spec); + + pexpr->AddRef(); + CExpression *pexprMotion = nullptr; + + // Generate appropriate motion based on required distribution type + switch (pdsRequired->Edt()) + { + case CDistributionSpec::EdtHashed: + { + // Required: Hashed distribution -> Generate HashDistribute Motion + if (GPOS_FTRACE(EopttraceDisableMotionHashDistribute)) + { + // Hash redistribute Motion is disabled, cannot satisfy requirement + pexpr->Release(); + return; + } + + CDistributionSpecHashed *pdsHashedRequired = + CDistributionSpecHashed::PdsConvert(pdsRequired); + pdsHashedRequired->AddRef(); + + if (fDuplicateHazard) + { + pdsHashedRequired->MarkDuplicateSensitive(); + } + + pexprMotion = GPOS_NEW(mp) CExpression( + mp, GPOS_NEW(mp) CPhysicalMotionHashDistribute(mp, pdsHashedRequired), pexpr); + break; + } + + case CDistributionSpec::EdtRandom: + { + // Required: Random distribution (segment-level, no worker parallelism) + // Need to convert WorkerRandom to plain Random + if (GPOS_FTRACE(EopttraceDisableMotionRandom)) + { + // Random Motion is disabled + pexpr->Release(); + return; + } + + // Create a Random distribution spec (not WorkerRandom) as the Motion target + // This converts worker-level parallelism to segment-level distribution + CDistributionSpecRandom *random_dist_spec = nullptr; + + if (fDuplicateHazard) + { + random_dist_spec = GPOS_NEW(mp) CDistributionSpecRandom(); + random_dist_spec->MarkDuplicateSensitive(); + } + else + { + // Use StrictRandom for actual redistribution motion + random_dist_spec = GPOS_NEW(mp) CDistributionSpecStrictRandom(); + } + + pexprMotion = GPOS_NEW(mp) CExpression( + mp, GPOS_NEW(mp) CPhysicalMotionRandom(mp, random_dist_spec), pexpr); + break; + } + case CDistributionSpec::EdtWorkerRandom: + { + // Required: WorkerRandom distribution -> Generate Random Motion + if (GPOS_FTRACE(EopttraceDisableMotionRandom)) + { + // Random Motion is disabled + pexpr->Release(); + return; + } + + CDistributionSpecWorkerRandom *random_dist_spec = + PdsCreateWorkerRandom(mp, m_ulWorkers, GPOS_NEW(mp) CDistributionSpecRandom()); + + if (fDuplicateHazard) + { + random_dist_spec->MarkDuplicateSensitive(); + } + + pexprMotion = GPOS_NEW(mp) CExpression( + mp, GPOS_NEW(mp) CPhysicalMotionRandom(mp, random_dist_spec), pexpr); + break; + } + + default: + { + // Fallback: cannot generate appropriate motion + pexpr->Release(); + return; + } + } + + // Add the generated motion to the enforcer array + if (nullptr != pexprMotion) + { + pdrgpexpr->Append(pexprMotion); + } +} + +//--------------------------------------------------------------------------- +// @function: +// CDistributionSpecWorkerRandom::OsPrint +// +// @doc: +// Print function +// +//--------------------------------------------------------------------------- +IOstream & +CDistributionSpecWorkerRandom::OsPrint(IOstream &os) const +{ + os << SzId() << "[workers:" << m_ulWorkers << "]"; + if (nullptr != m_pdsSegmentBase) + { + os << " base:"; + m_pdsSegmentBase->OsPrint(os); + } + if (IsDuplicateSensitive()) + { + os << " (duplicate sensitive)"; + } + return os; +} + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/src/base/Makefile b/src/backend/gporca/libgpopt/src/base/Makefile index 604fd3a6fa4..01f2bb0a02a 100644 --- a/src/backend/gporca/libgpopt/src/base/Makefile +++ b/src/backend/gporca/libgpopt/src/base/Makefile @@ -42,6 +42,7 @@ OBJS = CAutoOptCtxt.o \ CDistributionSpecStrictRandom.o \ CDistributionSpecStrictSingleton.o \ CDistributionSpecUniversal.o \ + CDistributionSpecWorkerRandom.o \ CDrvdProp.o \ CDrvdPropCtxt.o \ CDrvdPropCtxtPlan.o \ diff --git a/src/backend/gporca/libgpopt/src/operators/CLogicalGet.cpp b/src/backend/gporca/libgpopt/src/operators/CLogicalGet.cpp index ff93a28522c..8e6e3461d51 100644 --- a/src/backend/gporca/libgpopt/src/operators/CLogicalGet.cpp +++ b/src/backend/gporca/libgpopt/src/operators/CLogicalGet.cpp @@ -305,6 +305,8 @@ CLogicalGet::PxfsCandidates(CMemoryPool *mp) const CXformSet *xform_set = GPOS_NEW(mp) CXformSet(mp); (void) xform_set->ExchangeSet(CXform::ExfGet2TableScan); + // add parallel table scan + (void) xform_set->ExchangeSet(CXform::ExfGet2ParallelTableScan); return xform_set; } diff --git a/src/backend/gporca/libgpopt/src/operators/CPhysicalCTEProducer.cpp b/src/backend/gporca/libgpopt/src/operators/CPhysicalCTEProducer.cpp index c6be21f87cd..a13e89fda1e 100644 --- a/src/backend/gporca/libgpopt/src/operators/CPhysicalCTEProducer.cpp +++ b/src/backend/gporca/libgpopt/src/operators/CPhysicalCTEProducer.cpp @@ -14,6 +14,7 @@ #include "gpos/base.h" #include "gpopt/base/CCTEMap.h" +#include "gpopt/base/CDistributionSpecRandom.h" #include "gpopt/base/COptCtxt.h" #include "gpopt/operators/CExpression.h" #include "gpopt/operators/CExpressionHandle.h" diff --git a/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelTableScan.cpp b/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelTableScan.cpp new file mode 100644 index 00000000000..a70a35981c6 --- /dev/null +++ b/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelTableScan.cpp @@ -0,0 +1,284 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CPhysicalParallelTableScan.cpp + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/src/operators/CPhysicalParallelTableScan.cpp + * + *------------------------------------------------------------------------- + */ + +#include "gpopt/operators/CPhysicalParallelTableScan.h" + +#include "gpos/base.h" + +#include "gpopt/base/CDistributionSpec.h" +#include "gpopt/base/CDistributionSpecHashed.h" +#include "gpopt/base/CDistributionSpecRandom.h" +#include "gpopt/base/CDistributionSpecWorkerRandom.h" +#include "gpopt/base/CDistributionSpecSingleton.h" +#include "gpopt/base/CUtils.h" +#include "gpopt/base/CEnfdDistribution.h" +#include "gpopt/base/CEnfdRewindability.h" +#include "gpopt/base/COptimizationContext.h" +#include "gpopt/base/CRewindabilitySpec.h" +#include "gpopt/base/CDrvdPropPlan.h" +#include "gpopt/metadata/CName.h" +#include "gpopt/metadata/CTableDescriptor.h" +#include "gpopt/operators/CExpressionHandle.h" + +using namespace gpopt; + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::CPhysicalParallelTableScan +// +// @doc: +// ctor +// +//--------------------------------------------------------------------------- +CPhysicalParallelTableScan::CPhysicalParallelTableScan(CMemoryPool *mp) + : CPhysicalTableScan(mp, GPOS_NEW(mp) CName(GPOS_NEW(mp) CWStringConst(GPOS_WSZ_LIT("parallel_table"))), nullptr, nullptr), + m_ulParallelWorkers(1), + m_pdsWorkerDistribution(nullptr) +{ +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::CPhysicalParallelTableScan +// +// @doc: +// ctor +// +//--------------------------------------------------------------------------- +CPhysicalParallelTableScan::CPhysicalParallelTableScan(CMemoryPool *mp, + const CName *pnameAlias, + CTableDescriptor *ptabdesc, + CColRefArray *pdrgpcrOutput, + ULONG ulParallelWorkers) + : CPhysicalTableScan(mp, pnameAlias, ptabdesc, pdrgpcrOutput), + m_ulParallelWorkers(ulParallelWorkers), + m_pdsWorkerDistribution(nullptr) +{ + GPOS_ASSERT(ulParallelWorkers > 0); + GPOS_ASSERT(nullptr != m_pds); + // Create worker-level distribution based on table's segment distribution + if (ulParallelWorkers > 0 && nullptr != m_pds) + { + // Create worker-level random distribution using the table's distribution as base + // The base CPhysicalScan already sets up m_pds from the table descriptor + m_pdsWorkerDistribution = CDistributionSpecWorkerRandom::PdsCreateWorkerRandom(mp, ulParallelWorkers, m_pds); + } +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::~CPhysicalParallelTableScan +// +// @doc: +// dtor +// +//--------------------------------------------------------------------------- +CPhysicalParallelTableScan::~CPhysicalParallelTableScan() +{ + CRefCount::SafeRelease(m_pdsWorkerDistribution); +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::HashValue +// +// @doc: +// Combine pointer for table descriptor, parallel workers and Eop +// +//--------------------------------------------------------------------------- +ULONG +CPhysicalParallelTableScan::HashValue() const +{ + ULONG ulHash = gpos::CombineHashes(CPhysicalTableScan::HashValue(), + gpos::HashValue<ULONG>(&m_ulParallelWorkers)); + return ulHash; +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::Matches +// +// @doc: +// match operator +// +//--------------------------------------------------------------------------- +BOOL +CPhysicalParallelTableScan::Matches(COperator *pop) const +{ + if (Eopid() != pop->Eopid()) + { + return false; + } + + CPhysicalParallelTableScan *popParallelTableScan = + CPhysicalParallelTableScan::PopConvert(pop); + + return CPhysicalTableScan::Matches(pop) && + m_ulParallelWorkers == popParallelTableScan->UlParallelWorkers(); +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::OsPrint +// +// @doc: +// debug print +// +//--------------------------------------------------------------------------- +IOstream & +CPhysicalParallelTableScan::OsPrint(IOstream &os) const +{ + os << SzId() << " "; + + // alias of table as referenced in the query + m_pnameAlias->OsPrint(os); + + // actual name of table in catalog and columns + os << " ("; + m_ptabdesc->Name().OsPrint(os); + os << "), Columns: ["; + + CUtils::OsPrintDrgPcr(os, m_pdrgpcrOutput); + os << "], Workers: " << m_ulParallelWorkers; + + return os; +} + + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::PdsDerive +// +// @doc: +// Derive distribution for parallel table scan +// +//--------------------------------------------------------------------------- +CDistributionSpec * +CPhysicalParallelTableScan::PdsDerive(CMemoryPool *mp, CExpressionHandle &exprhdl) const +{ + // If we have a pre-computed worker distribution, use it + if (nullptr != m_pdsWorkerDistribution) + { + m_pdsWorkerDistribution->AddRef(); + return m_pdsWorkerDistribution; + } + + // Otherwise, derive from the base physical scan + // This uses the m_pds member from CPhysicalScan + return CPhysicalScan::PdsDerive(mp, exprhdl); +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::EpetDistribution +// +// @doc: +// Return the enforcing type for distribution property based on this +// operator +// +//--------------------------------------------------------------------------- +CEnfdProp::EPropEnforcingType +CPhysicalParallelTableScan::EpetDistribution(CExpressionHandle & /*exprhdl*/, + const CEnfdDistribution *ped) const +{ + GPOS_ASSERT(nullptr != ped); + + // First check if worker-level distribution can satisfy the requirement + // This is the primary distribution for parallel scans + if (nullptr != m_pdsWorkerDistribution && ped->FCompatible(m_pdsWorkerDistribution)) + { + return CEnfdProp::EpetUnnecessary; + } + + // Neither distribution satisfies the requirement + // Motion enforcement will be needed on the output + return CEnfdProp::EpetRequired; +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::EpetRewindability +// +// @doc: +// Return rewindability property enforcing type for this operator +// +//--------------------------------------------------------------------------- +CEnfdProp::EPropEnforcingType +CPhysicalParallelTableScan::EpetRewindability(CExpressionHandle &exprhdl, + const CEnfdRewindability *per) const +{ + GPOS_ASSERT(nullptr != per); + + // Get derived rewindability from this operator + CRewindabilitySpec *prs = CDrvdPropPlan::Pdpplan(exprhdl.Pdp())->Prs(); + + // Check if our derived rewindability satisfies the requirement + if (per->FCompatible(prs)) + { + // Our derived rewindability (ErtNone) satisfies the requirement + return CEnfdProp::EpetUnnecessary; + } + + // Cannot satisfy the rewindability requirement + // GPORCA will need to add an enforcer (e.g., Spool) + return CEnfdProp::EpetRequired; +} + +//--------------------------------------------------------------------------- +// @function: +// CPhysicalParallelTableScan::FValidContext +// +// @doc: +// Check if optimization contexts is valid; +// Reject if parent requires REWINDABLE (e.g., for NL Join inner child) +// because ParallelTableScan derives NONE (not rewindable) +// +//--------------------------------------------------------------------------- +BOOL +CPhysicalParallelTableScan::FValidContext(CMemoryPool *, + COptimizationContext *poc, + COptimizationContextArray *) const +{ + GPOS_ASSERT(nullptr != poc); + + CReqdPropPlan *prpp = poc->Prpp(); + CRewindabilitySpec *prsRequired = prpp->Per()->PrsRequired(); + + // If parent requires REWINDABLE or higher, reject + // ParallelTableScan can only provide ErtNone + if (prsRequired->IsOriginNLJoin()) + { + // Parent requires rewindability (e.g., NL Join inner child) + // but ParallelTableScan cannot provide it + // Reject this plan to avoid the assertion failure later + return false; + } + + return true; +} + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/src/operators/Makefile b/src/backend/gporca/libgpopt/src/operators/Makefile index faa0f9a70d6..ce5362e9049 100644 --- a/src/backend/gporca/libgpopt/src/operators/Makefile +++ b/src/backend/gporca/libgpopt/src/operators/Makefile @@ -123,6 +123,7 @@ OBJS = CExpression.o \ CPhysicalMotionRoutedDistribute.o \ CPhysicalDynamicForeignScan.o \ CPhysicalNLJoin.o \ + CPhysicalParallelTableScan.o \ CPhysicalParallelUnionAll.o \ CPhysicalPartitionSelector.o \ CPhysicalRightOuterHashJoin.o \ diff --git a/src/backend/gporca/libgpopt/src/optimizer/COptimizerConfig.cpp b/src/backend/gporca/libgpopt/src/optimizer/COptimizerConfig.cpp index e55a04862aa..a9486f33e09 100644 --- a/src/backend/gporca/libgpopt/src/optimizer/COptimizerConfig.cpp +++ b/src/backend/gporca/libgpopt/src/optimizer/COptimizerConfig.cpp @@ -33,14 +33,16 @@ COptimizerConfig::COptimizerConfig(CEnumeratorConfig *pec, CStatisticsConfig *stats_config, CCTEConfig *pcteconf, ICostModel *cost_model, CHint *phint, CPlanHint *pplanhint, - CWindowOids *pwindowoids) + CWindowOids *pwindowoids, + BOOL enable_parallel_plans) : m_enumerator_cfg(pec), m_stats_conf(stats_config), m_cte_conf(pcteconf), m_cost_model(cost_model), m_hint(phint), m_plan_hint(pplanhint), - m_window_oids(pwindowoids) + m_window_oids(pwindowoids), + m_create_parallel_plan(enable_parallel_plans) { GPOS_ASSERT(nullptr != pec); GPOS_ASSERT(nullptr != stats_config); @@ -85,7 +87,7 @@ COptimizerConfig::PoconfDefault(CMemoryPool *mp) CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), ICostModel::PcmDefault(mp), CHint::PhintDefault(mp), nullptr /* pplanhint */, - CWindowOids::GetWindowOids(mp)); + CWindowOids::GetWindowOids(mp), false /* enable_parallel_plans */); } //--------------------------------------------------------------------------- @@ -105,7 +107,7 @@ COptimizerConfig::PoconfDefault(CMemoryPool *mp, ICostModel *pcm) GPOS_NEW(mp) CEnumeratorConfig(mp, 0 /*plan_id*/, 0 /*ullSamples*/), CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), pcm, CHint::PhintDefault(mp), - nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp)); + nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp), false /* enable_parallel_plans */); } //--------------------------------------------------------------------------- diff --git a/src/backend/gporca/libgpopt/src/search/CGroup.cpp b/src/backend/gporca/libgpopt/src/search/CGroup.cpp index 18b4cf44e5b..02fd7fa9ccc 100644 --- a/src/backend/gporca/libgpopt/src/search/CGroup.cpp +++ b/src/backend/gporca/libgpopt/src/search/CGroup.cpp @@ -152,6 +152,7 @@ CGroup::SContextLink::Equals(const SContextLink *pclink1, //--------------------------------------------------------------------------- CGroup::CGroup(CMemoryPool *mp, BOOL fScalar) : m_mp(mp), + m_pmemo(nullptr), m_id(GPOPT_INVALID_GROUP_ID), m_fScalar(fScalar), m_pdrgpexprJoinKeysOuter(nullptr), @@ -534,6 +535,25 @@ CGroup::SetState(EState estNewState) } +//--------------------------------------------------------------------------- +// @function: +// CGroup::SetMemo +// +// @doc: +// Set containing memo reference; +// +//--------------------------------------------------------------------------- +void +CGroup::SetMemo(CMemo *pmemo) +{ + GPOS_ASSERT(nullptr != pmemo); + GPOS_ASSERT(nullptr == m_pmemo && + "Overwriting previously assigned memo reference"); + + m_pmemo = pmemo; +} + + void CGroup::SetJoinKeys(CExpressionArray *pdrgpexprOuter, CExpressionArray *pdrgpexprInner, diff --git a/src/backend/gporca/libgpopt/src/search/CMemo.cpp b/src/backend/gporca/libgpopt/src/search/CMemo.cpp index 5248300aa86..d92da7a9ed9 100644 --- a/src/backend/gporca/libgpopt/src/search/CMemo.cpp +++ b/src/backend/gporca/libgpopt/src/search/CMemo.cpp @@ -140,6 +140,7 @@ CMemo::Add( GPOS_ASSERT(nullptr != gp.PgexprFirst()); gp.SetId(id); + gp.SetMemo(this); gp.InitProperties(pdp); } diff --git a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp index 6bc89993008..51acac518be 100644 --- a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp +++ b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp @@ -66,6 +66,7 @@ #include "gpopt/operators/CPhysicalStreamAggDeduplicate.h" #include "gpopt/operators/CPhysicalTVF.h" #include "gpopt/operators/CPhysicalTableScan.h" +#include "gpopt/operators/CPhysicalParallelTableScan.h" #include "gpopt/operators/CPhysicalUnionAll.h" #include "gpopt/operators/CPredicateUtils.h" #include "gpopt/operators/CScalarArray.h" @@ -129,6 +130,7 @@ #include "naucrates/dxl/operators/CDXLPhysicalSplit.h" #include "naucrates/dxl/operators/CDXLPhysicalTVF.h" #include "naucrates/dxl/operators/CDXLPhysicalTableScan.h" +#include "naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h" #include "naucrates/dxl/operators/CDXLPhysicalWindow.h" #include "naucrates/dxl/operators/CDXLScalarAggref.h" #include "naucrates/dxl/operators/CDXLScalarArray.h" @@ -356,6 +358,7 @@ CTranslatorExprToDXL::CreateDXLNode(CExpression *pexpr, GPOS_ASSERT(nullptr != pexpr); ULONG ulOpId = (ULONG) pexpr->Pop()->Eopid(); if (COperator::EopPhysicalTableScan == ulOpId || + COperator::EopPhysicalParallelTableScan == ulOpId || COperator::EopPhysicalForeignScan == ulOpId) { CDXLNode *dxlnode = PdxlnTblScan( @@ -725,6 +728,13 @@ CTranslatorExprToDXL::PdxlnTblScan(CExpression *pexprTblScan, { pdxlopTS = GPOS_NEW(m_mp) CDXLPhysicalTableScan(m_mp, table_descr); } + else if (COperator::EopPhysicalParallelTableScan == op_id) + { + CPhysicalParallelTableScan *parallel_scan = + CPhysicalParallelTableScan::PopConvert(pexprTblScan->Pop()); + pdxlopTS = GPOS_NEW(m_mp) CDXLPhysicalParallelTableScan( + m_mp, table_descr, parallel_scan->UlParallelWorkers()); + } else { GPOS_ASSERT(COperator::EopPhysicalForeignScan == op_id); @@ -2570,6 +2580,7 @@ CTranslatorExprToDXL::PdxlnFromFilter(CExpression *pexprFilter, switch (eopidRelational) { case COperator::EopPhysicalTableScan: + case COperator::EopPhysicalParallelTableScan: case COperator::EopPhysicalForeignScan: { // if there is a structure of the form @@ -4409,6 +4420,7 @@ CTranslatorExprToDXL::PdxlnCorrelatedNLJoin( switch (pexprOuterChild->Pop()->Eopid()) { case COperator::EopPhysicalTableScan: + case COperator::EopPhysicalParallelTableScan: { dxl_properties->AddRef(); // create and return a table scan node @@ -4597,6 +4609,7 @@ UlIndexFilter(Edxlopid edxlopid) { case EdxlopPhysicalTableScan: case EdxlopPhysicalForeignScan: + case EdxlopPhysicalParallelTableScan: return EdxltsIndexFilter; case EdxlopPhysicalDynamicForeignScan: return EdxldfsIndexFilter; @@ -4652,6 +4665,7 @@ CTranslatorExprToDXL::PdxlnResultFromNLJoinOuter( case EdxlopPhysicalDynamicIndexScan: case EdxlopPhysicalDynamicBitmapTableScan: case EdxlopPhysicalResult: + case EdxlopPhysicalParallelTableScan: { // if the scalar join condition is a constant TRUE, just translate the child, no need to create an AND expression if (CTranslatorExprToDXLUtils::FScalarConstTrue(m_pmda, diff --git a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXLUtils.cpp b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXLUtils.cpp index cf497223553..7fa48972366 100644 --- a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXLUtils.cpp +++ b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXLUtils.cpp @@ -17,6 +17,7 @@ #include "gpopt/base/CConstraintDisjunction.h" #include "gpopt/base/CConstraintInterval.h" #include "gpopt/base/CConstraintNegation.h" +#include "gpopt/base/CDistributionSpecWorkerRandom.h" #include "gpopt/base/COptCtxt.h" #include "gpopt/exception.h" #include "gpopt/mdcache/CMDAccessorUtils.h" @@ -759,10 +760,11 @@ CTranslatorExprToDXLUtils::SetDirectDispatchInfo( // +--CScalarConst (5) if (CDistributionSpec::EdtHashed == pds->Edt() || - CDistributionSpec::EdtRandom == pds->Edt()) + CDistributionSpec::EdtRandom == pds->Edt() || + CDistributionSpec::EdtWorkerRandom == pds->Edt()) { // direct dispatch supported for scans over - // hash & random distributed tables + // hash, random & worker-random distributed tables for (ULONG i = 0; i < size; i++) { CExpression *pexprFilter = (*pexprFilterArray)[i]; @@ -816,6 +818,62 @@ CTranslatorExprToDXLUtils::SetDirectDispatchInfo( dxl_direct_dispatch_info = GetDXLDirectDispatchInfoRandDist( mp, md_accessor, pcrDistrCol, pcnstrDistrCol); } + else if (CDistributionSpec::EdtWorkerRandom == pds->Edt()) + { + CConstraint *pcnstr = ppc->Pcnstr(); + + CDistributionSpecWorkerRandom *pdsWorkerRandom = + CDistributionSpecWorkerRandom::PdsConvert(pds); + + // Get the base segment distribution for worker-random distribution + CDistributionSpec *pdsSegmentBase = pdsWorkerRandom->PdsSegmentBase(); + + if (nullptr == pdsSegmentBase) + { + // No base segment distribution available, cannot proceed with direct dispatch + continue; + } + + // Handle direct dispatch based on the base segment distribution type + if (CDistributionSpec::EdtHashed == pdsSegmentBase->Edt()) + { + // Base distribution is hashed - use hash distribution keys + CDistributionSpecHashed *pdsHashed = + CDistributionSpecHashed::PdsConvert(pdsSegmentBase); + CExpressionArray *pdrgpexprHashed = pdsHashed->Pdrgpexpr(); + + dxl_direct_dispatch_info = GetDXLDirectDispatchInfo( + mp, md_accessor, pdrgpexprHashed, pcnstr); + } + else if (CDistributionSpec::EdtRandom == pdsSegmentBase->Edt()) + { + // Base distribution is random - use gp_segment_id + CDistributionSpecRandom *pdsRandom = + CDistributionSpecRandom::PdsConvert(pdsSegmentBase); + + // Extracting GpSegmentID for base random distribution + const CColRef *pcrDistrCol = pdsRandom->GetGpSegmentId(); + + if (pcrDistrCol == nullptr) + { + // Direct Dispatch not feasible - no gp_segment_id available + continue; + } + + CConstraint *pcnstrDistrCol = pcnstr->Pcnstr(mp, pcrDistrCol); + + if (pcnstrDistrCol == nullptr) + { + // Direct Dispatch not feasible - no constraint on gp_segment_id + continue; + } + + dxl_direct_dispatch_info = GetDXLDirectDispatchInfoRandDist( + mp, md_accessor, pcrDistrCol, pcnstrDistrCol); + } + // Note: Other base distribution types (Singleton, Replicated, etc.) + // are not supported for direct dispatch in worker-random context + } if (nullptr != dxl_direct_dispatch_info) { diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp index e21c24511dd..7ffe39774dc 100644 --- a/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp +++ b/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp @@ -144,6 +144,7 @@ CXformFactory::Instantiate() Add(GPOS_NEW(m_mp) CXformExpandNAryJoinMinCard(m_mp)); Add(GPOS_NEW(m_mp) CXformExpandNAryJoinDP(m_mp)); Add(GPOS_NEW(m_mp) CXformGet2TableScan(m_mp)); + Add(GPOS_NEW(m_mp) CXformGet2ParallelTableScan(m_mp)); Add(GPOS_NEW(m_mp) CXformIndexGet2IndexScan(m_mp)); Add(GPOS_NEW(m_mp) CXformDynamicGet2DynamicTableScan(m_mp)); Add(GPOS_NEW(m_mp) CXformDynamicIndexGet2DynamicIndexScan(m_mp)); diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformGet2ParallelTableScan.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformGet2ParallelTableScan.cpp new file mode 100644 index 00000000000..8799661ce49 --- /dev/null +++ b/src/backend/gporca/libgpopt/src/xforms/CXformGet2ParallelTableScan.cpp @@ -0,0 +1,243 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CXformGet2ParallelTableScan.cpp + * + * IDENTIFICATION + * src/backend/gporca/libgpopt/src/xforms/CXformGet2ParallelTableScan.cpp + * + *------------------------------------------------------------------------- + */ + +#include "gpopt/xforms/CXformGet2ParallelTableScan.h" + +#include "gpos/base.h" + +#include "gpopt/base/COptCtxt.h" +#include "gpopt/hints/CHintUtils.h" +#include "gpopt/metadata/CTableDescriptor.h" +#include "gpopt/operators/CExpressionHandle.h" +#include "gpopt/operators/CLogicalGet.h" +#include "gpopt/operators/CPhysicalParallelTableScan.h" +#include "gpopt/optimizer/COptimizerConfig.h" +#include "naucrates/md/IMDRelation.h" +#include "gpopt/search/CGroupProxy.h" +#include "gpopt/search/CMemo.h" + + +// Use gpdbwrappers for parallel checks +extern int max_parallel_workers_per_gather; + +// Forward declarations for gpdbwrappers functions +namespace gpdb { + bool IsParallelModeOK(void); +} + +using namespace gpopt; + +//--------------------------------------------------------------------------- +// @function: +// CXformGet2ParallelTableScan::FHasParallelIncompatibleOps +// +// @doc: +// Check if memo contains logical operators that are incompatible +// with parallel execution (CTE, Dynamic scans, Foreign scans, etc.) +// +//--------------------------------------------------------------------------- +BOOL +CXformGet2ParallelTableScan::FHasParallelIncompatibleOps(CExpressionHandle &exprhdl) +{ + CGroupExpression *pgexprHandle = exprhdl.Pgexpr(); + if (nullptr == pgexprHandle) + { + return false; + } + + CGroup *pgroup = pgexprHandle->Pgroup(); + if (nullptr == pgroup) + { + return false; + } + + CMemo *pmemo = pgroup->Pmemo(); + if (nullptr == pmemo) + { + return false; + } + + // Iterate through all groups in memo to check for parallel-incompatible operations + const ULONG_PTR ulGroups = pmemo->UlpGroups(); + for (ULONG_PTR ul = 0; ul < ulGroups; ul++) + { + CGroup *pgroupCurrent = pmemo->Pgroup(ul); + if (nullptr == pgroupCurrent) + { + continue; + } + + // Check all group expressions in this group using CGroupProxy + CGroupProxy gp(pgroupCurrent); + CGroupExpression *pgexpr = gp.PgexprFirst(); + while (nullptr != pgexpr) + { + COperator::EOperatorId eopid = pgexpr->Pop()->Eopid(); + + // Check for CTE-related operators (incompatible with parallel execution) + if (COperator::EopLogicalCTEProducer == eopid || + COperator::EopLogicalCTEConsumer == eopid || + COperator::EopLogicalSequence == eopid || + COperator::EopLogicalSequenceProject == eopid) + { + return true; + } + + if (COperator::EopLogicalUnion == eopid || + COperator::EopLogicalUnionAll == eopid || + COperator::EopLogicalIntersect == eopid || + COperator::EopLogicalIntersectAll == eopid || + COperator::EopLogicalDifference == eopid || + COperator::EopLogicalDifferenceAll == eopid) + { + // Set operations are not supported in parallel plans + return true; + } + + pgexpr = gp.PgexprNext(pgexpr); + } + } + + return false; +} + +//--------------------------------------------------------------------------- +// @function: +// CXformGet2ParallelTableScan::CXformGet2ParallelTableScan +// +// @doc: +// Ctor +// +//--------------------------------------------------------------------------- +CXformGet2ParallelTableScan::CXformGet2ParallelTableScan(CMemoryPool *mp) + : CXformImplementation( + // pattern + GPOS_NEW(mp) CExpression(mp, GPOS_NEW(mp) CLogicalGet(mp))) +{ +} + +//--------------------------------------------------------------------------- +// @function: +// CXformGet2ParallelTableScan::Exfp +// +// @doc: +// Compute promise of xform based on GUC enable_parallel +// Uses unified parallel degree from max_parallel_workers_per_gather +// +//--------------------------------------------------------------------------- +CXform::EXformPromise +CXformGet2ParallelTableScan::Exfp(CExpressionHandle &exprhdl) const +{ + // Check if parallel plans are enabled in context and parallel processing is safe + if (!gpdb::IsParallelModeOK()) + { + return CXform::ExfpNone; + } + + // Check for parallel-incompatible operations that would conflict with parallel scans + if (FHasParallelIncompatibleOps(exprhdl)) + { + return CXform::ExfpNone; + } + + CLogicalGet *popGet = CLogicalGet::PopConvert(exprhdl.Pop()); + CTableDescriptor *ptabdesc = popGet->Ptabdesc(); + + // Don't use parallel scan for replicated tables + if (ptabdesc->GetRelDistribution() == IMDRelation::EreldistrReplicated || + ptabdesc->GetRelDistribution() == IMDRelation::EreldistrMasterOnly || + COptCtxt::PoctxtFromTLS()->HasReplicatedTables()) + { + //FIXME: Should we consider replicated tables. + return CXform::ExfpNone; + } + + // High promise for parallel scan when enabled + // All tables will use the same parallel degree from max_parallel_workers_per_gather + return CXform::ExfpHigh; +} + +//--------------------------------------------------------------------------- +// @function: +// CXformGet2ParallelTableScan::Transform +// +// @doc: +// Actual transformation +// +//--------------------------------------------------------------------------- +void +CXformGet2ParallelTableScan::Transform(CXformContext *pxfctxt, CXformResult *pxfres, + CExpression *pexpr) const +{ + GPOS_ASSERT(nullptr != pxfctxt); + GPOS_ASSERT(FPromising(pxfctxt->Pmp(), this, pexpr)); + GPOS_ASSERT(FCheckPattern(pexpr)); + + CLogicalGet *popGet = CLogicalGet::PopConvert(pexpr->Pop()); + + CMemoryPool *mp = pxfctxt->Pmp(); + + // create/extract components for alternative + CName *pname = GPOS_NEW(mp) CName(mp, popGet->Name()); + + CTableDescriptor *ptabdesc = popGet->Ptabdesc(); + ptabdesc->AddRef(); + + CColRefArray *pdrgpcrOutput = popGet->PdrgpcrOutput(); + GPOS_ASSERT(nullptr != pdrgpcrOutput); + pdrgpcrOutput->AddRef(); + + // Determine parallel workers degree + // Priority: table-level parallel_workers setting > GUC max_parallel_workers_per_gather > default + ULONG ulParallelWorkers = 2; + + // Check if table has a specific parallel_workers setting + CMDAccessor *md_accessor = COptCtxt::PoctxtFromTLS()->Pmda(); + const IMDRelation *pmdrel = md_accessor->RetrieveRel(ptabdesc->MDId()); + INT table_parallel_workers = pmdrel->ParallelWorkers(); + + if (table_parallel_workers > 0) + { + // Use table-level setting if explicitly configured + ulParallelWorkers = (ULONG)table_parallel_workers; + } + else if (max_parallel_workers_per_gather > 0) + { + // Fall back to GUC setting + ulParallelWorkers = (ULONG)max_parallel_workers_per_gather; + } + + // create alternative expression + CExpression *pexprAlt = GPOS_NEW(mp) CExpression( + mp, + GPOS_NEW(mp) CPhysicalParallelTableScan(mp, pname, ptabdesc, pdrgpcrOutput, ulParallelWorkers)); + + // add alternative to transformation result + pxfres->Add(pexprAlt); +} + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformGet2TableScan.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformGet2TableScan.cpp index 05e826aec92..d34068a3dfc 100644 --- a/src/backend/gporca/libgpopt/src/xforms/CXformGet2TableScan.cpp +++ b/src/backend/gporca/libgpopt/src/xforms/CXformGet2TableScan.cpp @@ -20,6 +20,10 @@ #include "gpopt/operators/CPhysicalTableScan.h" #include "gpopt/optimizer/COptimizerConfig.h" +namespace gpdb { + bool IsParallelModeOK(void); +} + using namespace gpopt; @@ -57,6 +61,13 @@ CXformGet2TableScan::Exfp(CExpressionHandle &exprhdl) const return CXform::ExfpNone; } + // If parallel processing is enabled, give lower priority to regular table scan + // to allow parallel table scan to take precedence + if (gpdb::IsParallelModeOK()) + { + return CXform::ExfpLow; + } + return CXform::ExfpHigh; } diff --git a/src/backend/gporca/libgpopt/src/xforms/Makefile b/src/backend/gporca/libgpopt/src/xforms/Makefile index 03f6293b36d..66da51e28bb 100644 --- a/src/backend/gporca/libgpopt/src/xforms/Makefile +++ b/src/backend/gporca/libgpopt/src/xforms/Makefile @@ -47,6 +47,7 @@ OBJS = CDecorrelator.o \ CXformGbAggDedup2StreamAggDedup.o \ CXformGbAggWithMDQA2Join.o \ CXformGet2TableScan.o \ + CXformGet2ParallelTableScan.o \ CXformImplementAssert.o \ CXformImplementBitmapTableGet.o \ CXformImplementCTEConsumer.o \ diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h index 8fb27307c36..e583e9974a3 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h @@ -125,6 +125,7 @@ enum Edxlopid EdxlopPhysicalValuesScan, EdxlopPhysicalProjection, EdxlopPhysicalTableScan, + EdxlopPhysicalParallelTableScan, EdxlopPhysicalBitmapTableScan, EdxlopPhysicalDynamicBitmapTableScan, EdxlopPhysicalForeignScan, diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h index ed27db129fe..568240a5c98 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h @@ -134,6 +134,10 @@ public: static CDXLPhysical *MakeDXLTblScan(CDXLMemoryManager *dxl_memory_manager, const Attributes &attrs); + // create a parallel table scan operator + static CDXLPhysical *MakeDXLParallelTblScan(CDXLMemoryManager *dxl_memory_manager, + const Attributes &attrs); + // create a result operator static CDXLPhysical *MakeDXLResult(CDXLMemoryManager *dxl_memory_manager); diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h new file mode 100644 index 00000000000..ff3afa68df0 --- /dev/null +++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h @@ -0,0 +1,104 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CDXLPhysicalParallelTableScan.h + * + * IDENTIFICATION + * src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h + * + *------------------------------------------------------------------------- + */ + +#ifndef GPDXL_CDXLPhysicalParallelTableScan_H +#define GPDXL_CDXLPhysicalParallelTableScan_H + +#include "gpos/base.h" + +#include "naucrates/dxl/operators/CDXLPhysicalTableScan.h" + +namespace gpdxl +{ +using namespace gpos; + +//--------------------------------------------------------------------------- +// @class: +// CDXLPhysicalParallelTableScan +// +// @doc: +// Class for representing DXL parallel table scan operators +// +//--------------------------------------------------------------------------- +class CDXLPhysicalParallelTableScan : public CDXLPhysicalTableScan +{ +private: + // number of parallel workers + ULONG m_ulParallelWorkers; + +public: + CDXLPhysicalParallelTableScan(const CDXLPhysicalParallelTableScan &) = delete; + + // ctor + CDXLPhysicalParallelTableScan(CMemoryPool *mp, CDXLTableDescr *table_descr, + ULONG ulParallelWorkers); + + // ctor with uninitialized table descriptor + CDXLPhysicalParallelTableScan(CMemoryPool *mp, ULONG ulParallelWorkers); + + // dtor + ~CDXLPhysicalParallelTableScan() override = default; + + // get operator type + Edxlopid GetDXLOperator() const override; + + // get operator name + const CWStringConst *GetOpNameStr() const override; + + // get number of parallel workers + ULONG UlParallelWorkers() const + { + return m_ulParallelWorkers; + } + + // serialize operator in DXL format + void SerializeToDXL(CXMLSerializer *xml_serializer, + const CDXLNode *dxlnode) const override; + + // conversion function + static CDXLPhysicalParallelTableScan * + Cast(CDXLOperator *dxl_op) + { + GPOS_ASSERT(nullptr != dxl_op); + GPOS_ASSERT(EdxlopPhysicalParallelTableScan == dxl_op->GetDXLOperator()); + + return dynamic_cast<CDXLPhysicalParallelTableScan *>(dxl_op); + } + +#ifdef GPOS_DEBUG + // checks whether the operator has valid structure, i.e. number and + // types of child nodes + void AssertValid(const CDXLNode *dxlnode, BOOL validate_children) const override; +#endif // GPOS_DEBUG + +}; // class CDXLPhysicalParallelTableScan + +} // namespace gpdxl + +#endif // !GPDXL_CDXLPhysicalParallelTableScan_H + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalTableScan.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalTableScan.h index a043e137614..aef1695612a 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalTableScan.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalTableScan.h @@ -76,7 +76,8 @@ public: { GPOS_ASSERT(nullptr != dxl_op); GPOS_ASSERT(EdxlopPhysicalTableScan == dxl_op->GetDXLOperator() || - EdxlopPhysicalForeignScan == dxl_op->GetDXLOperator()); + EdxlopPhysicalForeignScan == dxl_op->GetDXLOperator() || + EdxlopPhysicalParallelTableScan == dxl_op->GetDXLOperator()); return dynamic_cast<CDXLPhysicalTableScan *>(dxl_op); } diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h index 5a61801511a..9c5d89fe8ae 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h @@ -96,6 +96,7 @@ enum Edxltoken EdxltokenPhysical, EdxltokenPhysicalTableScan, + EdxltokenPhysicalParallelTableScan, EdxltokenPhysicalBitmapTableScan, EdxltokenPhysicalDynamicBitmapTableScan, EdxltokenPhysicalForeignScan, @@ -374,6 +375,7 @@ enum Edxltoken EdxltokenColNdvBySeg, EdxltokenParamId, + EdxltokenParallelWorkers, EdxltokenCtidColName, EdxltokenOidColName, diff --git a/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationCtasGPDB.h b/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationCtasGPDB.h index 780c1be5b11..195a17f9e94 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationCtasGPDB.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationCtasGPDB.h @@ -264,6 +264,12 @@ public: return 0; } + INT SegFileCount() const override + { + GPOS_ASSERT("Function should not be called for CTAS tables"); + return -1; + } + #ifdef GPOS_DEBUG // debug print of the metadata relation void DebugPrint(IOstream &os) const override; diff --git a/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationGPDB.h b/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationGPDB.h index f15b0f26ecc..ecb4bb4a4d4 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationGPDB.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/md/CMDRelationGPDB.h @@ -122,13 +122,19 @@ private: // rows CDouble m_rows; + // segment file count for AO/AOCO tables (-1 for non-AO tables) + INT m_seg_file_count; + + // parallel workers setting from table options (-1 if not set) + INT m_parallel_workers; + public: CMDRelationGPDB(const CMDRelationGPDB &) = delete; // ctor CMDRelationGPDB( CMemoryPool *mp, IMDId *mdid, CMDName *mdname, BOOL is_temp_table, - Erelstoragetype rel_storage_type, + Erelstoragetype rel_storage_type, Ereldistrpolicy rel_distr_policy, CMDColumnArray *mdcol_array, ULongPtrArray *distr_col_array, IMdIdArray *distr_opfamilies, ULongPtrArray *partition_cols_array, CharPtrArray *str_part_types_array, @@ -242,6 +248,16 @@ public: CDouble Rows() const override; + INT SegFileCount() const override; + + // Set segment file count for AO/AOCO tables + void SetSegFileCount(INT seg_file_count); + + INT ParallelWorkers() const override; + + // Set parallel workers from table options + void SetParallelWorkers(INT parallel_workers); + #ifdef GPOS_DEBUG // debug print of the metadata relation void DebugPrint(IOstream &os) const override; diff --git a/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelation.h b/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelation.h index 036af05597a..b5e7a8f22ee 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelation.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelation.h @@ -203,6 +203,12 @@ public: // rows virtual CDouble Rows() const = 0; + + // segment file count for AO/AOCO tables (returns -1 for non-AO tables) + virtual INT SegFileCount() const = 0; + + // parallel workers setting from table options (returns -1 if not set) + virtual INT ParallelWorkers() const = 0; }; // common structure over relation and external relation metadata for index info diff --git a/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelationCtas.h b/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelationCtas.h index 18aa6b8e30c..ae805c9cc17 100644 --- a/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelationCtas.h +++ b/src/backend/gporca/libnaucrates/include/naucrates/md/IMDRelationCtas.h @@ -108,6 +108,13 @@ public: // CTAS storage options virtual CDXLCtasStorageOptions *GetDxlCtasStorageOption() const = 0; + + // parallel workers - CTAS tables don't have this setting yet + INT + ParallelWorkers() const override + { + return -1; // not set + } }; } // namespace gpmd diff --git a/src/backend/gporca/libnaucrates/src/md/CMDRelationGPDB.cpp b/src/backend/gporca/libnaucrates/src/md/CMDRelationGPDB.cpp index 03f52c3fdd0..dceb24d0c55 100644 --- a/src/backend/gporca/libnaucrates/src/md/CMDRelationGPDB.cpp +++ b/src/backend/gporca/libnaucrates/src/md/CMDRelationGPDB.cpp @@ -62,7 +62,9 @@ CMDRelationGPDB::CMDRelationGPDB( m_colpos_nondrop_colpos_map(nullptr), m_attrno_nondrop_col_pos_map(nullptr), m_nondrop_col_pos_array(nullptr), - m_rows(rows) + m_rows(rows), + m_seg_file_count(-1), + m_parallel_workers(-1) { GPOS_ASSERT(mdid->IsValid()); GPOS_ASSERT(nullptr != mdcol_array); @@ -595,6 +597,46 @@ CMDRelationGPDB::Rows() const return m_rows; } +INT +CMDRelationGPDB::SegFileCount() const +{ + return m_seg_file_count; +} + +//--------------------------------------------------------------------------- +// @function: +// CMDRelationGPDB::SetSegFileCount +// +// @doc: +// Set segment file count for AO/AOCO tables +// +//--------------------------------------------------------------------------- +void +CMDRelationGPDB::SetSegFileCount(INT seg_file_count) +{ + m_seg_file_count = seg_file_count; +} + +INT +CMDRelationGPDB::ParallelWorkers() const +{ + return m_parallel_workers; +} + +//--------------------------------------------------------------------------- +// @function: +// CMDRelationGPDB::SetParallelWorkers +// +// @doc: +// Set parallel workers from table options +// +//--------------------------------------------------------------------------- +void +CMDRelationGPDB::SetParallelWorkers(INT parallel_workers) +{ + m_parallel_workers = parallel_workers; +} + //--------------------------------------------------------------------------- // @function: // CMDRelationGPDB::Serialize diff --git a/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp b/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp index 8b037b833f0..a4d4914b731 100644 --- a/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp +++ b/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp @@ -28,6 +28,7 @@ #include "naucrates/dxl/operators/CDXLDatumStatsDoubleMappable.h" #include "naucrates/dxl/operators/CDXLDatumStatsLintMappable.h" #include "naucrates/dxl/operators/CDXLLogicalJoin.h" +#include "naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h" #include "naucrates/dxl/operators/CDXLPhysicalAgg.h" #include "naucrates/dxl/operators/CDXLPhysicalAppend.h" #include "naucrates/dxl/operators/CDXLPhysicalBroadcastMotion.h" @@ -101,6 +102,31 @@ CDXLOperatorFactory::MakeDXLTblScan(CDXLMemoryManager *dxl_memory_manager, return GPOS_NEW(mp) CDXLPhysicalTableScan(mp); } +//--------------------------------------------------------------------------- +// @function: +// CDXLOperatorFactory::MakeDXLParallelTblScan +// +// @doc: +// Construct a parallel table scan operator +// +//--------------------------------------------------------------------------- +CDXLPhysical * +CDXLOperatorFactory::MakeDXLParallelTblScan(CDXLMemoryManager *dxl_memory_manager, + const Attributes &attrs) +{ + // get the memory pool from the memory manager + CMemoryPool *mp = dxl_memory_manager->Pmp(); + + // extract number of parallel workers + const XMLCh *parallel_workers_xml = ExtractAttrValue(attrs, EdxltokenParallelWorkers, + EdxltokenPhysicalParallelTableScan); + ULONG ulParallelWorkers = CDXLOperatorFactory::ConvertAttrValueToUlong( + dxl_memory_manager, parallel_workers_xml, EdxltokenParallelWorkers, + EdxltokenPhysicalParallelTableScan); + + return GPOS_NEW(mp) CDXLPhysicalParallelTableScan(mp, ulParallelWorkers); +} + //--------------------------------------------------------------------------- // @function: // CDXLOperatorFactory::MakeDXLResult diff --git a/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelTableScan.cpp b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelTableScan.cpp new file mode 100644 index 00000000000..a96657031e9 --- /dev/null +++ b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelTableScan.cpp @@ -0,0 +1,162 @@ +/*------------------------------------------------------------------------- + * + * 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. + * + * CDXLPhysicalParallelTableScan.cpp + * + * IDENTIFICATION + * src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelTableScan.cpp + * + *------------------------------------------------------------------------- + */ + +#include "naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h" + +#include "naucrates/dxl/operators/CDXLNode.h" +#include "naucrates/dxl/xml/CXMLSerializer.h" + +using namespace gpos; +using namespace gpdxl; + +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::CDXLPhysicalParallelTableScan +// +// @doc: +// Constructor +// +//--------------------------------------------------------------------------- +CDXLPhysicalParallelTableScan::CDXLPhysicalParallelTableScan(CMemoryPool *mp, + CDXLTableDescr *table_descr, + ULONG ulParallelWorkers) + : CDXLPhysicalTableScan(mp, table_descr), + m_ulParallelWorkers(ulParallelWorkers) +{ + GPOS_ASSERT(ulParallelWorkers > 0); +} + +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::CDXLPhysicalParallelTableScan +// +// @doc: +// Constructor with uninitialized table descriptor +// +//--------------------------------------------------------------------------- +CDXLPhysicalParallelTableScan::CDXLPhysicalParallelTableScan(CMemoryPool *mp, + ULONG ulParallelWorkers) + : CDXLPhysicalTableScan(mp), + m_ulParallelWorkers(ulParallelWorkers) +{ + GPOS_ASSERT(ulParallelWorkers > 0); +} + +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::GetDXLOperator +// +// @doc: +// Operator type +// +//--------------------------------------------------------------------------- +Edxlopid +CDXLPhysicalParallelTableScan::GetDXLOperator() const +{ + return EdxlopPhysicalParallelTableScan; +} + +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::GetOpNameStr +// +// @doc: +// Operator name +// +//--------------------------------------------------------------------------- +const CWStringConst * +CDXLPhysicalParallelTableScan::GetOpNameStr() const +{ + return CDXLTokens::GetDXLTokenStr(EdxltokenPhysicalParallelTableScan); +} + +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::SerializeToDXL +// +// @doc: +// Serialize operator in DXL format +// +//--------------------------------------------------------------------------- +void +CDXLPhysicalParallelTableScan::SerializeToDXL(CXMLSerializer *xml_serializer, + const CDXLNode *dxlnode) const +{ + const CWStringConst *element_name = GetOpNameStr(); + xml_serializer->OpenElement(CDXLTokens::GetDXLTokenStr(EdxltokenNamespacePrefix), + element_name); + + // serialize parallel workers attribute + xml_serializer->AddAttribute(CDXLTokens::GetDXLTokenStr(EdxltokenParallelWorkers), + m_ulParallelWorkers); + + // serialize properties + dxlnode->SerializePropertiesToDXL(xml_serializer); + + // serialize projection list and filter + dxlnode->SerializeChildrenToDXL(xml_serializer); + + xml_serializer->CloseElement(CDXLTokens::GetDXLTokenStr(EdxltokenNamespacePrefix), + element_name); +} + +#ifdef GPOS_DEBUG +//--------------------------------------------------------------------------- +// @function: +// CDXLPhysicalParallelTableScan::AssertValid +// +// @doc: +// Checks whether operator node is well-structured +// +//--------------------------------------------------------------------------- +void +CDXLPhysicalParallelTableScan::AssertValid(const CDXLNode *dxlnode, + BOOL validate_children) const +{ + // assert proj list and filter are valid + CDXLPhysical::AssertValid(dxlnode, validate_children); + + // parallel table scan has only 2 children (proj list and filter) + GPOS_ASSERT(2 == dxlnode->Arity()); + + CDXLNode *proj_list_dxlnode = (*dxlnode)[0]; // First child is projection list + CDXLNode *filter_dxlnode = (*dxlnode)[1]; // Second child is filter + + GPOS_ASSERT(EdxlopScalarProjectList == + proj_list_dxlnode->GetOperator()->GetDXLOperator()); + GPOS_ASSERT(EdxlopScalarFilter == + filter_dxlnode->GetOperator()->GetDXLOperator()); + + if (validate_children) + { + proj_list_dxlnode->GetOperator()->AssertValid(proj_list_dxlnode, validate_children); + filter_dxlnode->GetOperator()->AssertValid(filter_dxlnode, validate_children); + } +} +#endif // GPOS_DEBUG + +// EOF \ No newline at end of file diff --git a/src/backend/gporca/libnaucrates/src/operators/Makefile b/src/backend/gporca/libnaucrates/src/operators/Makefile index 418bd5a0fd2..1d78142feab 100644 --- a/src/backend/gporca/libnaucrates/src/operators/Makefile +++ b/src/backend/gporca/libnaucrates/src/operators/Makefile @@ -85,6 +85,7 @@ OBJS = CDXLColDescr.o \ CDXLPhysicalSplit.o \ CDXLPhysicalTVF.o \ CDXLPhysicalTableScan.o \ + CDXLPhysicalParallelTableScan.o \ CDXLPhysicalValuesScan.o \ CDXLPhysicalWindow.o \ CDXLProperties.o \ diff --git a/src/backend/gporca/libnaucrates/src/parser/CParseHandlerOptimizerConfig.cpp b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerOptimizerConfig.cpp index 3b64d409bfe..49b4d786b55 100644 --- a/src/backend/gporca/libnaucrates/src/parser/CParseHandlerOptimizerConfig.cpp +++ b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerOptimizerConfig.cpp @@ -273,7 +273,7 @@ CParseHandlerOptimizerConfig::EndElement(const XMLCh *const, // element_uri, } m_optimizer_config = GPOS_NEW(m_mp) COptimizerConfig( - pec, stats_config, pcteconfig, pcm, phint, pplanhint, pwindowoidsGPDB); + pec, stats_config, pcteconfig, pcm, phint, pplanhint, pwindowoidsGPDB, false); CParseHandlerTraceFlags *pphTraceFlags = dynamic_cast<CParseHandlerTraceFlags *>((*this)[this->Length() - 1]); diff --git a/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp b/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp index 0f59d689a05..79fcc9f7d6a 100644 --- a/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp +++ b/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp @@ -112,6 +112,7 @@ CDXLTokens::Init(CMemoryPool *mp) {EdxltokenPhysical, GPOS_WSZ_LIT("OpPhysical")}, {EdxltokenPhysicalTableScan, GPOS_WSZ_LIT("TableScan")}, + {EdxltokenPhysicalParallelTableScan, GPOS_WSZ_LIT("ParallelTableScan")}, {EdxltokenPhysicalBitmapTableScan, GPOS_WSZ_LIT("BitmapTableScan")}, {EdxltokenPhysicalDynamicBitmapTableScan, GPOS_WSZ_LIT("DynamicBitmapTableScan")}, @@ -420,6 +421,7 @@ CDXLTokens::Init(CMemoryPool *mp) {EdxltokenColNdvBySeg, GPOS_WSZ_LIT("NdvBySeg")}, {EdxltokenParamId, GPOS_WSZ_LIT("ParamId")}, + {EdxltokenParallelWorkers, GPOS_WSZ_LIT("ParallelWorkers")}, {EdxltokenCtidColName, GPOS_WSZ_LIT("ctid")}, {EdxltokenOidColName, GPOS_WSZ_LIT("oid")}, diff --git a/src/backend/gporca/server/src/startup/main.cpp b/src/backend/gporca/server/src/startup/main.cpp index 29be9dc9994..9418fa94abb 100644 --- a/src/backend/gporca/server/src/startup/main.cpp +++ b/src/backend/gporca/server/src/startup/main.cpp @@ -381,6 +381,24 @@ PvExec(void *pv) return nullptr; } +//--------------------------------------------------------------------------- +// Stub implementations for PostgreSQL symbols required by GPORCA +// These are needed for standalone testing without linking to PostgreSQL +//--------------------------------------------------------------------------- + +// Stub for parallel mode check +namespace gpdb { + bool IsParallelModeOK(void) + { + // For unittest, we enable parallel mode by default + return true; + } +} + +// Stub GUC variables for parallel execution +int max_parallel_workers_per_gather = 2; +bool enable_parallel = true; +double parallel_setup_cost = 1000.0; //--------------------------------------------------------------------------- // @function: diff --git a/src/backend/gporca/server/src/unittest/CTestUtils.cpp b/src/backend/gporca/server/src/unittest/CTestUtils.cpp index bca99766fda..5ce83305b32 100644 --- a/src/backend/gporca/server/src/unittest/CTestUtils.cpp +++ b/src/backend/gporca/server/src/unittest/CTestUtils.cpp @@ -3539,7 +3539,7 @@ CTestUtils::EresSamplePlans(const CHAR *rgszFileNames[], ULONG ulTests, CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), ICostModel::PcmDefault(mp), CHint::PhintDefault(mp), nullptr, - CWindowOids::GetWindowOids(mp)); + CWindowOids::GetWindowOids(mp), false); } else { @@ -3678,7 +3678,7 @@ CTestUtils::EresCheckPlans(const CHAR *rgszFileNames[], ULONG ulTests, CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), ICostModel::PcmDefault(mp), CHint::PhintDefault(mp), nullptr, - CWindowOids::GetWindowOids(mp)); + CWindowOids::GetWindowOids(mp), false); } else { diff --git a/src/backend/gporca/server/src/unittest/dxl/statistics/CCardinalityTestUtils.cpp b/src/backend/gporca/server/src/unittest/dxl/statistics/CCardinalityTestUtils.cpp index f95ecb14ee4..24da702044c 100644 --- a/src/backend/gporca/server/src/unittest/dxl/statistics/CCardinalityTestUtils.cpp +++ b/src/backend/gporca/server/src/unittest/dxl/statistics/CCardinalityTestUtils.cpp @@ -112,7 +112,8 @@ CCardinalityTestUtils::PhistInt4Remain(CMemoryPool *mp, ULONG num_of_buckets, } return GPOS_NEW(mp) CHistogram(mp, histogram_buckets, true, null_freq, - num_NDV_remain, freq_remaining); + num_NDV_remain, freq_remaining, + num_NDV_remain /* distinct_by_segs */); } // helper function to generate an example int histogram diff --git a/src/backend/gporca/server/src/unittest/dxl/statistics/CHistogramTest.cpp b/src/backend/gporca/server/src/unittest/dxl/statistics/CHistogramTest.cpp index b6eb14a8378..900e16a3f7d 100644 --- a/src/backend/gporca/server/src/unittest/dxl/statistics/CHistogramTest.cpp +++ b/src/backend/gporca/server/src/unittest/dxl/statistics/CHistogramTest.cpp @@ -268,7 +268,8 @@ CHistogramTest::PhistExampleInt4Remain(CMemoryPool *mp) return GPOS_NEW(mp) CHistogram(mp, histogram_buckets, true, 0.1 /*null_freq*/, - 2.0 /*distinct_remaining*/, 0.4 /*freq_remaining*/); + 2.0 /*distinct_remaining*/, 0.4 /*freq_remaining*/, + 2.0 /*distinct_by_segs*/); } // basis skew test diff --git a/src/backend/gporca/server/src/unittest/dxl/statistics/CStatisticsTest.cpp b/src/backend/gporca/server/src/unittest/dxl/statistics/CStatisticsTest.cpp index e596add7f0d..6b0345b98e6 100644 --- a/src/backend/gporca/server/src/unittest/dxl/statistics/CStatisticsTest.cpp +++ b/src/backend/gporca/server/src/unittest/dxl/statistics/CStatisticsTest.cpp @@ -504,7 +504,7 @@ CStatisticsTest::EresUnittest_CStatisticsBasic() ULongPtrArray *aggs = GPOS_NEW(mp) ULongPtrArray(mp); CStatistics *pstats4 = CGroupByStatsProcessor::CalcGroupByStats( - mp, stats, GCs, aggs, nullptr /*keys*/); + mp, stats, GCs, aggs, nullptr /*keys*/, false /*is_partial*/); GPOS_TRACE(GPOS_WSZ_LIT("pstats4 = stats group by")); CCardinalityTestUtils::PrintStats(mp, pstats4); diff --git a/src/backend/gporca/server/src/unittest/gpopt/minidump/CICGTest.cpp b/src/backend/gporca/server/src/unittest/gpopt/minidump/CICGTest.cpp index 58d3971cfcf..facd4126ea2 100644 --- a/src/backend/gporca/server/src/unittest/gpopt/minidump/CICGTest.cpp +++ b/src/backend/gporca/server/src/unittest/gpopt/minidump/CICGTest.cpp @@ -303,7 +303,7 @@ CICGTest::EresUnittest_NegativeIndexApplyTests() CEnumeratorConfig::GetEnumeratorCfg(mp, 0 /*plan_id*/), CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), pcm, CHint::PhintDefault(mp), - nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp)); + nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp), false); CDXLNode *pdxlnPlan = CMinidumperUtils::PdxlnExecuteMinidump( mp, rgszNegativeIndexApplyFileNames[ul], GPOPT_TEST_SEGMENTS /*ulSegments*/, 1 /*ulSessionId*/, diff --git a/src/backend/gporca/server/src/unittest/gpopt/minidump/CMiniDumperDXLTest.cpp b/src/backend/gporca/server/src/unittest/gpopt/minidump/CMiniDumperDXLTest.cpp index f046a185f29..b088aa5f6c5 100644 --- a/src/backend/gporca/server/src/unittest/gpopt/minidump/CMiniDumperDXLTest.cpp +++ b/src/backend/gporca/server/src/unittest/gpopt/minidump/CMiniDumperDXLTest.cpp @@ -127,7 +127,7 @@ CMiniDumperDXLTest::EresUnittest_Basic() CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), ICostModel::PcmDefault(mp), CHint::PhintDefault(mp), nullptr /* pplanhint */, - CWindowOids::GetWindowOids(mp)); + CWindowOids::GetWindowOids(mp), false); // setup opt ctx CAutoOptCtxt aoc(mp, &mda, nullptr, /* pceeval */ diff --git a/src/backend/gporca/server/src/unittest/gpopt/minidump/CMissingStatsTest.cpp b/src/backend/gporca/server/src/unittest/gpopt/minidump/CMissingStatsTest.cpp index a2d362aa5a2..6dc10137187 100644 --- a/src/backend/gporca/server/src/unittest/gpopt/minidump/CMissingStatsTest.cpp +++ b/src/backend/gporca/server/src/unittest/gpopt/minidump/CMissingStatsTest.cpp @@ -88,7 +88,7 @@ CMissingStatsTest::EresUnittest_RunTests() CEnumeratorConfig::GetEnumeratorCfg(mp, 0 /*plan_id*/), CStatisticsConfig::PstatsconfDefault(mp), CCTEConfig::PcteconfDefault(mp), pcm, CHint::PhintDefault(mp), - nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp)); + nullptr /* pplanhint */, CWindowOids::GetWindowOids(mp), false); SMissingStatsTestCase testCase = rgtc[ul]; CDXLNode *pdxlnPlan = CMinidumperUtils::PdxlnExecuteMinidump( diff --git a/src/backend/gporca/server/src/unittest/gpopt/xforms/CXformTest.cpp b/src/backend/gporca/server/src/unittest/gpopt/xforms/CXformTest.cpp index 910a10ad2ab..ba8e554bf08 100644 --- a/src/backend/gporca/server/src/unittest/gpopt/xforms/CXformTest.cpp +++ b/src/backend/gporca/server/src/unittest/gpopt/xforms/CXformTest.cpp @@ -172,7 +172,7 @@ CXformTest::EresUnittest_ApplyXforms_CTE() pdrgpexpr->Append(pexprConsumer); COptCtxt::PoctxtFromTLS()->Pcteinfo()->IncrementConsumers(ulCTEId); - COptCtxt::PoctxtFromTLS()->Pcteinfo()->AddCTEConsumer(pexprNewConsumer); + COptCtxt::PoctxtFromTLS()->Pcteinfo()->AddCTEConsumer(pexprConsumer); pexprConsumer->AddRef(); CExpression *pexprSelect = diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index 9c7021d4cb1..7129ab3fab3 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -327,6 +327,27 @@ planner(Query *parse, const char *query_string, int cursorOptions, optimizer_options = palloc(sizeof(OptimizerOptions)); optimizer_options->create_vectorization_plan = false; + + /* + * Set parallel plan creation based on PostgreSQL planner's parallel safety checks. + * This synchronizes with the same conditions used in standard_planner(). + */ + if ((cursorOptions & CURSOR_OPT_PARALLEL_OK) != 0 && + IsUnderPostmaster && + parse->commandType == CMD_SELECT && + !parse->hasModifyingCTE && + max_parallel_workers_per_gather > 0 && + !IsParallelWorker()) + { + /* All cheap tests pass, check query tree for parallel safety */ + char maxParallelHazard = max_parallel_hazard(parse); + optimizer_options->create_parallel_plan = (maxParallelHazard != PROPARALLEL_UNSAFE); + } + else + { + /* Skip the query tree scan, assume unsafe */ + optimizer_options->create_parallel_plan = false; + } if (planner_hook) { if (gp_log_optimization_time) diff --git a/src/include/gpopt/gpdbwrappers.h b/src/include/gpopt/gpdbwrappers.h index 9ef53169599..01d7eaa8cce 100644 --- a/src/include/gpopt/gpdbwrappers.h +++ b/src/include/gpopt/gpdbwrappers.h @@ -720,6 +720,9 @@ List *GetMergeJoinOpFamilies(Oid opno); // get the OID of base elementtype fora given typid Oid GetBaseType(Oid typid); +// check if parallel mode is OK (comprehensive check) +bool IsParallelModeOK(void); + // returns the result of evaluating 'expr' as an Expr. Caller keeps ownership of 'expr' // and takes ownership of the result Expr *EvaluateExpr(Expr *expr, Oid result_type, int32 typmod); @@ -771,6 +774,8 @@ List *GetRelChildIndexes(Oid reloid); Oid GetForeignServerId(Oid reloid); +int16 GetAppendOnlySegmentFilesCount(Relation rel); + void GPDBLockRelationOid(Oid reloid, int lockmode); char *GetRelFdwName(Oid reloid); diff --git a/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h b/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h index 625ed5cd0a3..3cd5fc9d638 100644 --- a/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h +++ b/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h @@ -213,6 +213,13 @@ private: ctxt_translation_prev_siblings // translation contexts of previous siblings ); + // translate DXL parallel table scan node into a parallel SeqScan node + Plan *TranslateDXLParallelTblScan( + const CDXLNode *tbl_scan_dxlnode, CDXLTranslateContext *output_context, + CDXLTranslationContextArray * + ctxt_translation_prev_siblings // translation contexts of previous siblings + ); + // translate DXL index scan node into a IndexScan node Plan *TranslateDXLIndexScan( const CDXLNode *index_scan_dxlnode, @@ -657,6 +664,10 @@ private: // fill the aggno and transno for the aggnode static void TranslateAggFillInfo(CContextDXLToPlStmt *context, Aggref *aggref); + + // extract parallel workers from DXL node tree + static ULONG ExtractParallelWorkersFromDXL(const CDXLNode *dxlnode); + }; } // namespace gpdxl diff --git a/src/include/gpopt/utils/COptTasks.h b/src/include/gpopt/utils/COptTasks.h index bed8f36d51a..cc96e63739e 100644 --- a/src/include/gpopt/utils/COptTasks.h +++ b/src/include/gpopt/utils/COptTasks.h @@ -90,6 +90,9 @@ struct SOptContext // is serializing a plan to DXL required ? BOOL m_should_serialize_plan_dxl{false}; + // should generate parallel plans ? + BOOL m_create_parallel_plan{false}; + // did the optimizer fail unexpectedly? BOOL m_is_unexpected_failure{false}; @@ -129,7 +132,8 @@ private: // create optimizer configuration object static COptimizerConfig *CreateOptimizerConfig(CMemoryPool *mp, ICostModel *cost_model, - CPlanHint *plan_hints); + CPlanHint *plan_hints, + BOOL enable_parallel_plans = false); // optimize a query to a physical DXL static void *OptimizeTask(void *ptr); diff --git a/src/include/optimizer/orcaopt.h b/src/include/optimizer/orcaopt.h index 4a045b632c0..6c1f3075bc4 100644 --- a/src/include/optimizer/orcaopt.h +++ b/src/include/optimizer/orcaopt.h @@ -33,6 +33,7 @@ typedef struct OptimizerOptions { bool create_vectorization_plan; + bool create_parallel_plan; } OptimizerOptions; diff --git a/src/test/regress/GNUmakefile b/src/test/regress/GNUmakefile index 9bc8b67591e..60f8d491efb 100644 --- a/src/test/regress/GNUmakefile +++ b/src/test/regress/GNUmakefile @@ -223,6 +223,12 @@ installcheck-cbdb-parallel: all tablespace-setup twophase_pqexecparams $(pg_regress_installcheck) $(REGRESS_OPTS) --schedule=$(srcdir)/parallel_schedule --schedule=$(srcdir)/greenplum_schedule --max-connections=5 $(EXTRA_TESTS) --exclude-tests explain \ ) +installcheck-orca-parallel: all tablespace-setup twophase_pqexecparams + ( \ + export PGOPTIONS='-c optimizer=on -c enable_parallel=true -c min_parallel_table_scan_size=0 -c min_parallel_index_scan_size=0 -c parallel_setup_cost=0'; \ + $(pg_regress_installcheck) $(REGRESS_OPTS) --exclude-file=$(srcdir)/excluded_tests.conf --schedule=$(srcdir)/parallel_schedule --schedule=$(srcdir)/greenplum_schedule --max-connections=4 --ignore-plans $(EXTRA_TESTS) \ + ) + installcheck-tests: all $(pg_regress_installcheck) $(REGRESS_OPTS) $(TESTS) $(EXTRA_TESTS) diff --git a/src/test/regress/excluded_tests.conf b/src/test/regress/excluded_tests.conf new file mode 100644 index 00000000000..e29197946d4 --- /dev/null +++ b/src/test/regress/excluded_tests.conf @@ -0,0 +1,14 @@ +brin_bloom +join_hash +explain +memoize +cte_prune +tuplesort +gporca +brin_ao +brin_aocs +direct_dispatch +bfv_dd +bfv_dd_multicolumn +planhints +rowhints diff --git a/src/test/regress/expected/brin.out b/src/test/regress/expected/brin.out index 20807f31997..2a0d670300a 100644 --- a/src/test/regress/expected/brin.out +++ b/src/test/regress/expected/brin.out @@ -323,6 +323,7 @@ BEGIN SET enable_bitmapscan = 1; SET optimizer_enable_tablescan = 0; SET optimizer_enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; is_planner_plan := false; diff --git a/src/test/regress/expected/brin_bloom.out b/src/test/regress/expected/brin_bloom.out index 241847a2707..e636c9ba4ae 100644 --- a/src/test/regress/expected/brin_bloom.out +++ b/src/test/regress/expected/brin_bloom.out @@ -230,6 +230,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_bloom WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/expected/brin_bloom_optimizer.out b/src/test/regress/expected/brin_bloom_optimizer.out index 65622f51b4b..2fb93d2819f 100644 --- a/src/test/regress/expected/brin_bloom_optimizer.out +++ b/src/test/regress/expected/brin_bloom_optimizer.out @@ -230,6 +230,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_bloom WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/expected/brin_multi.out b/src/test/regress/expected/brin_multi.out index 51277fdb887..037e331f8ba 100644 --- a/src/test/regress/expected/brin_multi.out +++ b/src/test/regress/expected/brin_multi.out @@ -260,6 +260,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_multi WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/expected/brin_multi_optimizer.out b/src/test/regress/expected/brin_multi_optimizer.out index 39100bfeb80..ff372c1c66e 100644 --- a/src/test/regress/expected/brin_multi_optimizer.out +++ b/src/test/regress/expected/brin_multi_optimizer.out @@ -260,6 +260,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_multi WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/expected/brin_multi_optimizer_1.out b/src/test/regress/expected/brin_multi_optimizer_1.out index d995e7888d2..da6ed1d404a 100644 --- a/src/test/regress/expected/brin_multi_optimizer_1.out +++ b/src/test/regress/expected/brin_multi_optimizer_1.out @@ -260,6 +260,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_multi WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/expected/brin_optimizer.out b/src/test/regress/expected/brin_optimizer.out index 37d3d21fb10..b749771b08e 100644 --- a/src/test/regress/expected/brin_optimizer.out +++ b/src/test/regress/expected/brin_optimizer.out @@ -323,6 +323,7 @@ BEGIN SET enable_bitmapscan = 1; SET optimizer_enable_tablescan = 0; SET optimizer_enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; is_planner_plan := false; diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out index 0bc2aaa9f6e..f71a81ef9cf 100644 --- a/src/test/regress/expected/create_index.out +++ b/src/test/regress/expected/create_index.out @@ -1705,7 +1705,7 @@ DROP TABLE syscol_table; -- -- Tests for IS NULL/IS NOT NULL with b-tree indexes -- -CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek; +CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek DISTRIBUTED BY (unique1); INSERT INTO onek_with_null (unique1,unique2) VALUES (NULL, -1), (NULL, NULL); CREATE UNIQUE INDEX onek_nulltest ON onek_with_null (unique2,unique1); SET enable_seqscan = OFF; diff --git a/src/test/regress/expected/create_index_optimizer.out b/src/test/regress/expected/create_index_optimizer.out index aca6fbb1332..c133678d152 100644 --- a/src/test/regress/expected/create_index_optimizer.out +++ b/src/test/regress/expected/create_index_optimizer.out @@ -1733,7 +1733,7 @@ DROP TABLE syscol_table; -- -- Tests for IS NULL/IS NOT NULL with b-tree indexes -- -CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek; +CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek DISTRIBUTED BY (unique1); INSERT INTO onek_with_null (unique1,unique2) VALUES (NULL, -1), (NULL, NULL); CREATE UNIQUE INDEX onek_nulltest ON onek_with_null (unique2,unique1); SET enable_seqscan = OFF; diff --git a/src/test/regress/expected/rpt_optimizer.out b/src/test/regress/expected/rpt_optimizer.out index 24ee61ff642..a07e2c8f8dc 100644 --- a/src/test/regress/expected/rpt_optimizer.out +++ b/src/test/regress/expected/rpt_optimizer.out @@ -933,16 +933,17 @@ explain (costs off) select a from t_replicate_volatile union all select * from n -- CTAS explain (costs off) create table rpt_ctas as select random() from generate_series(1, 10) distributed replicated; -QUERY PLAN -___________ + QUERY PLAN +---------------------------------------------- Result -> Broadcast Motion 1:3 (slice1) -> Function Scan on generate_series -GP_IGNORE:(4 rows) + Optimizer: GPORCA +(4 rows) explain (costs off) create table rpt_ctas as select a from generate_series(1, 10) a group by a having sum(a) > random() distributed replicated; -QUERY PLAN -___________ + QUERY PLAN +------------------------------------------------------------------------------- Result -> Broadcast Motion 3:3 (slice1; segments: 3) -> Result @@ -951,8 +952,8 @@ ___________ Group Key: generate_series -> Result -> Function Scan on generate_series - Optimizer: Pivotal Optimizer (GPORCA) -GP_IGNORE:(9 rows) + Optimizer: GPORCA +(9 rows) -- update & delete explain (costs off) update t_replicate_volatile set a = 1 where b > random(); diff --git a/src/test/regress/expected/tuplesort.out b/src/test/regress/expected/tuplesort.out index 47b5a8ddb21..e6cd26efdb1 100644 --- a/src/test/regress/expected/tuplesort.out +++ b/src/test/regress/expected/tuplesort.out @@ -236,7 +236,7 @@ CLUSTER abbrev_abort_uuids USING abbrev_abort_uuids__abort_increasing_idx; -- head SELECT id, abort_increasing, abort_decreasing, noabort_increasing, noabort_decreasing FROM abbrev_abort_uuids -ORDER BY ctid LIMIT 5; +ORDER BY ctid, id LIMIT 5; id | abort_increasing | abort_decreasing | noabort_increasing | noabort_decreasing ----+--------------------------------------+--------------------------------------+--------------------------------------+-------------------------------------- 2 | 00000000-0000-0000-0000-000000000001 | 00000000-0000-0000-0000-000000019999 | 00000001-0000-0000-0000-000000000001 | 00009990-0000-0000-0000-000000019999 diff --git a/src/test/regress/expected/tuplesort_optimizer.out b/src/test/regress/expected/tuplesort_optimizer.out index d62e0a0e71b..4eaabea61b5 100644 --- a/src/test/regress/expected/tuplesort_optimizer.out +++ b/src/test/regress/expected/tuplesort_optimizer.out @@ -232,7 +232,7 @@ CLUSTER abbrev_abort_uuids USING abbrev_abort_uuids__abort_increasing_idx; -- head SELECT id, abort_increasing, abort_decreasing, noabort_increasing, noabort_decreasing FROM abbrev_abort_uuids -ORDER BY ctid LIMIT 5; +ORDER BY ctid, id LIMIT 5; id | abort_increasing | abort_decreasing | noabort_increasing | noabort_decreasing ----+--------------------------------------+--------------------------------------+--------------------------------------+-------------------------------------- 2 | 00000000-0000-0000-0000-000000000001 | 00000000-0000-0000-0000-000000019999 | 00000001-0000-0000-0000-000000000001 | 00009990-0000-0000-0000-000000019999 diff --git a/src/test/regress/expected/workfile/hashjoin_spill.out b/src/test/regress/expected/workfile/hashjoin_spill.out index 5633e670954..053f457294b 100644 --- a/src/test/regress/expected/workfile/hashjoin_spill.out +++ b/src/test/regress/expected/workfile/hashjoin_spill.out @@ -37,6 +37,8 @@ HINT: The 'DISTRIBUTED BY' clause determines the distribution of data. Make sur insert into test_hj_spill SELECT i,i,i%1000,i,i,i,i,i from (select generate_series(1, nsegments * 15000) as i from (select count(*) as nsegments from gp_segment_configuration where role='p' and content >= 0) foo) bar; +-- Collect statistics to ensure ORCA generates correct execution plan +ANALYZE test_hj_spill; SET statement_mem=1024; set gp_resqueue_print_operator_memory_limits=on; set gp_workfile_compression = on; diff --git a/src/test/regress/sql/brin.sql b/src/test/regress/sql/brin.sql index 9a446efdc82..fade709eaa4 100644 --- a/src/test/regress/sql/brin.sql +++ b/src/test/regress/sql/brin.sql @@ -330,6 +330,7 @@ BEGIN SET enable_bitmapscan = 1; SET optimizer_enable_tablescan = 0; SET optimizer_enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; is_planner_plan := false; diff --git a/src/test/regress/sql/brin_bloom.sql b/src/test/regress/sql/brin_bloom.sql index 84572fdc2fb..558ca5730b7 100644 --- a/src/test/regress/sql/brin_bloom.sql +++ b/src/test/regress/sql/brin_bloom.sql @@ -231,6 +231,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_bloom WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/sql/brin_multi.sql b/src/test/regress/sql/brin_multi.sql index 9deb8d2573d..66028aaf1d5 100644 --- a/src/test/regress/sql/brin_multi.sql +++ b/src/test/regress/sql/brin_multi.sql @@ -266,6 +266,7 @@ BEGIN -- run the query using the brin index SET enable_seqscan = 0; SET enable_bitmapscan = 1; + SET enable_parallel = 0; plan_ok := false; FOR plan_line IN EXECUTE format($y$EXPLAIN SELECT array_agg(ctid) FROM brintest_multi WHERE %s $y$, cond) LOOP diff --git a/src/test/regress/sql/create_index.sql b/src/test/regress/sql/create_index.sql index ac5526ea9b1..b01c11aaf4f 100644 --- a/src/test/regress/sql/create_index.sql +++ b/src/test/regress/sql/create_index.sql @@ -649,7 +649,7 @@ DROP TABLE syscol_table; -- Tests for IS NULL/IS NOT NULL with b-tree indexes -- -CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek; +CREATE TABLE onek_with_null AS SELECT unique1, unique2 FROM onek DISTRIBUTED BY (unique1); INSERT INTO onek_with_null (unique1,unique2) VALUES (NULL, -1), (NULL, NULL); CREATE UNIQUE INDEX onek_nulltest ON onek_with_null (unique2,unique1); diff --git a/src/test/regress/sql/tuplesort.sql b/src/test/regress/sql/tuplesort.sql index 45c7c027cd8..fa5f87c7ec9 100644 --- a/src/test/regress/sql/tuplesort.sql +++ b/src/test/regress/sql/tuplesort.sql @@ -95,7 +95,7 @@ CLUSTER abbrev_abort_uuids USING abbrev_abort_uuids__abort_increasing_idx; -- head SELECT id, abort_increasing, abort_decreasing, noabort_increasing, noabort_decreasing FROM abbrev_abort_uuids -ORDER BY ctid LIMIT 5; +ORDER BY ctid, id LIMIT 5; -- tail SELECT id, abort_increasing, abort_decreasing, noabort_increasing, noabort_decreasing diff --git a/src/test/regress/sql/workfile/hashjoin_spill.sql b/src/test/regress/sql/workfile/hashjoin_spill.sql index 08e30501ef2..7166842dae5 100644 --- a/src/test/regress/sql/workfile/hashjoin_spill.sql +++ b/src/test/regress/sql/workfile/hashjoin_spill.sql @@ -40,6 +40,10 @@ CREATE TABLE test_hj_spill (i1 int, i2 int, i3 int, i4 int, i5 int, i6 int, i7 i insert into test_hj_spill SELECT i,i,i%1000,i,i,i,i,i from (select generate_series(1, nsegments * 15000) as i from (select count(*) as nsegments from gp_segment_configuration where role='p' and content >= 0) foo) bar; + +-- Collect statistics to ensure ORCA generates correct execution plan +ANALYZE test_hj_spill; + SET statement_mem=1024; set gp_resqueue_print_operator_memory_limits=on; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
