You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by yj...@apache.org on 2026/10/19 03:18:29 UTC
(cloudberry) 01/05: Implement parallel append in ORCA.
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 af408272ba18252831cab8e7567751ee5ce83a93
Author: zhangwenchao <65...@qq.com>
AuthorDate: Mon Nov 17 15:14:34 2025 +0800
Implement parallel append in ORCA.
Orca support parallel append plan, when open optimizer_enable_parallel_append guc,
orca will generate parallel append path and choose the best cost between parallel
append and dynamic scan. Parallel append only support parallel scan while scan can
not be added to its child node. In this way, we can realize the parallel execution
plan for partitioned tables.
---
src/backend/gpopt/config/CConfigParamMapping.cpp | 4 +
.../gpopt/translate/CTranslatorDXLToPlStmt.cpp | 143 ++++++++
.../gporca/libgpdbcost/src/CCostModelGPDB.cpp | 29 +-
.../libgpopt/include/gpopt/operators/COperator.h | 3 +
.../gpopt/operators/CPhysicalAppendTableScan.h | 99 ++++++
.../operators/CPhysicalParallelAppendTableScan.h | 143 ++++++++
.../include/gpopt/translate/CTranslatorExprToDXL.h | 29 ++
.../gporca/libgpopt/include/gpopt/xforms/CXform.h | 2 +
...leScan.h => CXformDynamicGet2AppendTableScan.h} | 37 +--
... => CXformDynamicGet2ParallelAppendTableScan.h} | 37 +--
.../gpopt/xforms/CXformGet2ParallelTableScan.h | 4 +-
.../gporca/libgpopt/include/gpopt/xforms/xforms.h | 2 +
.../gporca/libgpopt/src/base/CQueryContext.cpp | 4 +-
.../libgpopt/src/operators/CLogicalDynamicGet.cpp | 2 +
.../src/operators/CPhysicalAppendTableScan.cpp | 111 +++++++
.../operators/CPhysicalParallelAppendTableScan.cpp | 271 +++++++++++++++
src/backend/gporca/libgpopt/src/operators/Makefile | 2 +
.../src/translate/CTranslatorExprToDXL.cpp | 368 +++++++++++++++++++++
.../xforms/CXformDynamicGet2AppendTableScan.cpp | 130 ++++++++
.../CXformDynamicGet2ParallelAppendTableScan.cpp | 184 +++++++++++
.../gporca/libgpopt/src/xforms/CXformFactory.cpp | 2 +
src/backend/gporca/libgpopt/src/xforms/Makefile | 2 +
.../include/naucrates/dxl/operators/CDXLOperator.h | 1 +
.../naucrates/dxl/operators/CDXLOperatorFactory.h | 4 +
.../naucrates/dxl/operators/CDXLPhysicalAppend.h | 2 +-
.../dxl/operators/CDXLPhysicalParallelAppend.h | 143 ++++++++
.../naucrates/dxl/parser/CParseHandlerFactory.h | 5 +
.../dxl/parser/CParseHandlerParallelAppend.h | 84 +++++
.../include/naucrates/dxl/parser/parsehandlers.h | 1 +
.../include/naucrates/dxl/xml/dxltokens.h | 1 +
.../include/naucrates/traceflags/traceflags.h | 3 +
.../src/operators/CDXLOperatorFactory.cpp | 43 +++
.../src/operators/CDXLPhysicalAppend.cpp | 6 +-
...alAppend.cpp => CDXLPhysicalParallelAppend.cpp} | 118 +++----
.../gporca/libnaucrates/src/operators/Makefile | 1 +
.../src/parser/CParseHandlerFactory.cpp | 11 +
.../src/parser/CParseHandlerParallelAppend.cpp | 237 +++++++++++++
.../gporca/libnaucrates/src/parser/Makefile | 1 +
.../gporca/libnaucrates/src/xml/dxltokens.cpp | 1 +
src/backend/utils/misc/guc_gp.c | 11 +
.../gpopt/translate/CTranslatorDXLToPlStmt.h | 6 +
src/include/utils/guc.h | 1 +
src/include/utils/unsync_guc_name.h | 1 +
src/test/regress/GNUmakefile | 2 +-
src/test/regress/excluded_tests.conf | 3 +
src/test/regress/expected/orca_parallel_append.out | 354 ++++++++++++++++++++
src/test/regress/greenplum_schedule | 3 +
src/test/regress/sql/orca_parallel_append.sql | 125 +++++++
48 files changed, 2660 insertions(+), 116 deletions(-)
diff --git a/src/backend/gpopt/config/CConfigParamMapping.cpp b/src/backend/gpopt/config/CConfigParamMapping.cpp
index 5bd9091f604..3f617cc423e 100644
--- a/src/backend/gpopt/config/CConfigParamMapping.cpp
+++ b/src/backend/gpopt/config/CConfigParamMapping.cpp
@@ -284,6 +284,10 @@ CConfigParamMapping::SConfigMappingElem CConfigParamMapping::m_elements[] = {
GPOS_WSZ_LIT(
"Enable Eager Agg transform for pushing aggregate below an innerjoin.")},
+ {EopttraceEnableParallelAppendScan, &optimizer_enable_parallel_append,
+ false, // m_negate_param
+ GPOS_WSZ_LIT("Enable parallel append for scan/bitmap/index scan in partition tables.")},
+
{EopttraceDisableOrderedAgg, &optimizer_enable_orderedagg,
true, // m_negate_param
GPOS_WSZ_LIT("Disable ordered aggregate plans.")},
diff --git a/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp b/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp
index 4b9ae09b062..c2d65e43077 100644
--- a/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp
+++ b/src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp
@@ -59,6 +59,7 @@ extern "C" {
#include "naucrates/dxl/operators/CDXLNode.h"
#include "naucrates/dxl/operators/CDXLPhysicalAgg.h"
#include "naucrates/dxl/operators/CDXLPhysicalAppend.h"
+#include "naucrates/dxl/operators/CDXLPhysicalParallelAppend.h"
#include "naucrates/dxl/operators/CDXLPhysicalAssert.h"
#include "naucrates/dxl/operators/CDXLPhysicalBitmapTableScan.h"
#include "naucrates/dxl/operators/CDXLPhysicalCTAS.h"
@@ -443,6 +444,12 @@ CTranslatorDXLToPlStmt::TranslateDXLOperatorToPlan(
ctxt_translation_prev_siblings);
break;
}
+ case EdxlopPhysicalParallelAppend:
+ {
+ plan = TranslateDXLParallelAppend(dxlnode, output_context,
+ ctxt_translation_prev_siblings);
+ break;
+ }
case EdxlopPhysicalMaterialize:
{
plan = TranslateDXLMaterialize(dxlnode, output_context,
@@ -4574,6 +4581,142 @@ CTranslatorDXLToPlStmt::TranslateDXLAppend(
return (Plan *) append;
}
+//---------------------------------------------------------------------------
+// @function:
+// CTranslatorDXLToPlStmt::TranslateDXLParallelAppend
+//
+// @doc:
+// Translates a DXL parallel append node into a parallel Append node
+Plan *
+CTranslatorDXLToPlStmt::TranslateDXLParallelAppend(
+ const CDXLNode *append_dxlnode, CDXLTranslateContext *output_context,
+ CDXLTranslationContextArray *ctxt_translation_prev_siblings)
+{
+ CDXLPhysicalParallelAppend *phy_parallel_append_dxlop =
+ CDXLPhysicalParallelAppend::Cast(append_dxlnode->GetOperator());
+
+ ULONG parallel_workers = phy_parallel_append_dxlop->UlParallelWorkers();
+
+ // create append plan node
+ Append *append = MakeNode(Append);
+ append->first_partial_plan = 0;
+
+ Plan *plan = &(append->plan);
+ plan->plan_node_id = m_dxl_to_plstmt_context->GetNextPlanId();
+
+ // Set parallel execution flags
+ plan->parallel_aware = true;
+ plan->parallel_safe = true;
+ plan->parallel = (int) parallel_workers;
+
+ // translate operator costs
+ TranslatePlanCosts(append_dxlnode, plan);
+
+ const ULONG arity = append_dxlnode->Arity();
+ GPOS_ASSERT(EdxlappendIndexFirstChild < arity);
+ append->appendplans = NIL;
+
+ // translate table descriptor into a range table entry
+ CDXLPhysicalParallelAppend *phy_append_dxlop =
+ CDXLPhysicalParallelAppend::Cast(append_dxlnode->GetOperator());
+
+ // If this append was create from a DynamicTableScan node in ORCA, it will
+ // contain the table descriptor of the root partitioned table. Add that to
+ // the range table in the PlStmt.
+ if (phy_append_dxlop->GetScanId() != gpos::ulong_max)
+ {
+ GPOS_ASSERT(nullptr != phy_append_dxlop->GetDXLTableDesc());
+
+ // translation context for column mappings in the base relation
+ CDXLTranslateContextBaseTable base_table_context(m_mp);
+
+ (void) ProcessDXLTblDescr(phy_append_dxlop->GetDXLTableDesc(),
+ &base_table_context);
+
+ OID oid_type =
+ CMDIdGPDB::CastMdid(m_md_accessor->PtMDType<IMDTypeInt4>()->MDId())
+ ->Oid();
+ append->join_prune_paramids =
+ TranslateJoinPruneParamids(phy_append_dxlop->GetSelectorIds(),
+ oid_type, m_dxl_to_plstmt_context);
+ }
+
+ // translate children
+ CDXLTranslateContext child_context(m_mp, false,
+ output_context->GetColIdToParamIdMap());
+ for (ULONG ul = EdxlappendIndexFirstChild; ul < arity; ul++)
+ {
+ CDXLNode *child_dxlnode = (*append_dxlnode)[ul];
+
+ Plan *child_plan = TranslateDXLOperatorToPlan(
+ child_dxlnode, &child_context, ctxt_translation_prev_siblings);
+
+ GPOS_ASSERT(nullptr != child_plan && "child plan cannot be NULL");
+
+ append->appendplans = gpdb::LAppend(append->appendplans, child_plan);
+ }
+
+ CDXLNode *project_list_dxlnode = (*append_dxlnode)[EdxlappendIndexProjList];
+ CDXLNode *filter_dxlnode = (*append_dxlnode)[EdxlappendIndexFilter];
+
+ plan->targetlist = NIL;
+ const ULONG length = project_list_dxlnode->Arity();
+ for (ULONG ul = 0; ul < length; ++ul)
+ {
+ CDXLNode *proj_elem_dxlnode = (*project_list_dxlnode)[ul];
+ GPOS_ASSERT(EdxlopScalarProjectElem ==
+ proj_elem_dxlnode->GetOperator()->GetDXLOperator());
+
+ CDXLScalarProjElem *sc_proj_elem_dxlop =
+ CDXLScalarProjElem::Cast(proj_elem_dxlnode->GetOperator());
+ GPOS_ASSERT(1 == proj_elem_dxlnode->Arity());
+
+ // translate proj element expression
+ CDXLNode *expr_dxlnode = (*proj_elem_dxlnode)[0];
+ CDXLScalarIdent *sc_ident_dxlop =
+ CDXLScalarIdent::Cast(expr_dxlnode->GetOperator());
+
+ Index idxVarno = OUTER_VAR;
+ AttrNumber attno = (AttrNumber)(ul + 1);
+
+ Var *var = gpdb::MakeVar(
+ idxVarno, attno,
+ CMDIdGPDB::CastMdid(sc_ident_dxlop->MdidType())->Oid(),
+ sc_ident_dxlop->TypeModifier(),
+ 0 // varlevelsup
+ );
+
+ TargetEntry *target_entry = MakeNode(TargetEntry);
+ target_entry->expr = (Expr *) var;
+ target_entry->resname =
+ CTranslatorUtils::CreateMultiByteCharStringFromWCString(
+ sc_proj_elem_dxlop->GetMdNameAlias()->GetMDName()->GetBuffer());
+ target_entry->resno = attno;
+
+ // add column mapping to output translation context
+ output_context->InsertMapping(sc_proj_elem_dxlop->Id(), target_entry);
+
+ plan->targetlist = gpdb::LAppend(plan->targetlist, target_entry);
+ }
+
+ CDXLTranslationContextArray *child_contexts =
+ GPOS_NEW(m_mp) CDXLTranslationContextArray(m_mp);
+ child_contexts->Append(output_context);
+
+ // translate filter
+ plan->qual = TranslateDXLFilterToQual(
+ filter_dxlnode,
+ nullptr, // translate context for the base table
+ child_contexts, output_context);
+
+ SetParamIds(plan);
+
+ // cleanup
+ child_contexts->Release();
+
+ return (Plan *) append;
+};
+
//---------------------------------------------------------------------------
// @function:
// CTranslatorDXLToPlStmt::TranslateDXLMaterialize
diff --git a/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp b/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp
index 575f6473ff0..f4dad6c59df 100644
--- a/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp
+++ b/src/backend/gporca/libgpdbcost/src/CCostModelGPDB.cpp
@@ -29,6 +29,7 @@
#include "gpopt/operators/CPhysicalIndexOnlyScan.h"
#include "gpopt/operators/CPhysicalIndexScan.h"
#include "gpopt/operators/CPhysicalParallelTableScan.h"
+#include "gpopt/operators/CPhysicalParallelAppendTableScan.h"
#include "gpopt/operators/CPhysicalMotion.h"
#include "gpopt/operators/CPhysicalMotionBroadcast.h"
#include "gpopt/operators/CPhysicalPartitionSelector.h"
@@ -2385,7 +2386,9 @@ CCostModelGPDB::CostScan(CMemoryPool *, // mp
COperator::EopPhysicalDynamicTableScan == op_id ||
COperator::EopPhysicalForeignScan == op_id ||
COperator::EopPhysicalDynamicForeignScan == op_id ||
- COperator::EopPhysicalParallelTableScan == op_id);
+ COperator::EopPhysicalParallelTableScan == op_id ||
+ COperator::EopPhysicalAppendTableScan == op_id ||
+ COperator::EopPhysicalParallelAppendTableScan == op_id);
const CDouble dInitScan =
pcmgpdb->GetCostModelParams()
@@ -2408,6 +2411,8 @@ CCostModelGPDB::CostScan(CMemoryPool *, // mp
case COperator::EopPhysicalForeignScan:
case COperator::EopPhysicalDynamicForeignScan:
case COperator::EopPhysicalParallelTableScan:
+ case COperator::EopPhysicalAppendTableScan:
+ case COperator::EopPhysicalParallelAppendTableScan:
// 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,
@@ -2440,12 +2445,22 @@ CCostModelGPDB::CostParallelTableScan(CMemoryPool *mp,
GPOS_ASSERT(nullptr != pci);
COperator *pop = exprhdl.Pop();
- GPOS_ASSERT(COperator::EopPhysicalParallelTableScan == pop->Eopid());
+ GPOS_ASSERT(COperator::EopPhysicalParallelTableScan == pop->Eopid() ||
+ COperator::EopPhysicalParallelAppendTableScan == pop->Eopid());
- // Get the parallel table scan operator
- CPhysicalParallelTableScan *popParallelScan =
- CPhysicalParallelTableScan::PopConvert(pop);
- ULONG ulWorkers = popParallelScan->UlParallelWorkers();
+ // Get the parallel table scan operatora
+ ULONG ulWorkers = 0;
+
+ if (COperator::EopPhysicalParallelTableScan == pop->Eopid())
+ {
+ CPhysicalParallelTableScan *popParallelScan = CPhysicalParallelTableScan::PopConvert(pop);
+ ulWorkers = popParallelScan->UlParallelWorkers();
+ }
+ else if(COperator::EopPhysicalParallelAppendTableScan == pop->Eopid())
+ {
+ CPhysicalParallelAppendTableScan *popParallelScan = CPhysicalParallelAppendTableScan::PopConvert(pop);
+ ulWorkers = popParallelScan->UlParallelWorkers();
+ }
// If only 1 worker, use regular scan cost
if (ulWorkers <= 1)
@@ -2609,11 +2624,13 @@ CCostModelGPDB::Cost(
case COperator::EopPhysicalDynamicTableScan:
case COperator::EopPhysicalForeignScan:
case COperator::EopPhysicalDynamicForeignScan:
+ case COperator::EopPhysicalAppendTableScan:
{
return CostScan(m_mp, exprhdl, this, pci);
}
case COperator::EopPhysicalParallelTableScan:
+ case COperator::EopPhysicalParallelAppendTableScan:
{
return CostParallelTableScan(m_mp, exprhdl, this, pci);
}
diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h b/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h
index 5d10a1cee99..df540377214 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/operators/COperator.h
@@ -217,6 +217,9 @@ public:
EopPhysicalHashSequenceProject,
EopPhysicalDynamicIndexScan,
+ EopPhysicalAppendTableScan,
+ EopPhysicalParallelAppendTableScan,
+
EopPhysicalInnerHashJoin,
EopPhysicalLeftOuterHashJoin,
EopPhysicalLeftSemiHashJoin,
diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalAppendTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalAppendTableScan.h
new file mode 100644
index 00000000000..a321f8ffc64
--- /dev/null
+++ b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalAppendTableScan.h
@@ -0,0 +1,99 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalAppendTableScan.h
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalAppendTableScan.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef GPOPT_CPhysicalAppendTableScan_H
+#define GPOPT_CPhysicalAppendTableScan_H
+
+#include "gpos/base.h"
+
+#include "gpopt/operators/CPhysicalDynamicScan.h"
+
+namespace gpopt
+{
+//---------------------------------------------------------------------------
+// @class:
+// CPhysicalAppendTableScan
+//
+// @doc:
+// Append Table scan operator
+//
+//---------------------------------------------------------------------------
+class CPhysicalAppendTableScan : public CPhysicalDynamicScan
+{
+private:
+public:
+ CPhysicalAppendTableScan(const CPhysicalAppendTableScan &) = delete;
+
+ // ctors
+ CPhysicalAppendTableScan(CMemoryPool *mp, const CName *pnameAlias,
+ CTableDescriptor *ptabdesc, ULONG ulOriginOpId,
+ ULONG scan_id, CColRefArray *pdrgpcrOutput,
+ CColRef2dArray *pdrgpdrgpcrParts,
+ IMdIdArray *partition_mdids,
+ ColRefToUlongMapArray *root_col_mapping_per_part);
+
+ // ident accessors
+ EOperatorId
+ Eopid() const override
+ {
+ return EopPhysicalAppendTableScan;
+ }
+
+ // return a string for operator name
+ const CHAR *
+ SzId() const override
+ {
+ return "CPhysicalAppendTableScan";
+ }
+
+ // match function
+ BOOL Matches(COperator *) const override;
+
+ // statistics derivation during costing
+ IStatistics *PstatsDerive(CMemoryPool *mp, CExpressionHandle &exprhdl,
+ CReqdPropPlan *prpplan,
+ IStatisticsArray *stats_ctxt) const override;
+
+ // conversion function
+ static CPhysicalAppendTableScan *
+ PopConvert(COperator *pop)
+ {
+ GPOS_ASSERT(nullptr != pop);
+ GPOS_ASSERT(EopPhysicalAppendTableScan == pop->Eopid());
+
+ return dynamic_cast<CPhysicalAppendTableScan *>(pop);
+ }
+
+ CPartitionPropagationSpec *PppsDerive(
+ CMemoryPool *mp, CExpressionHandle &exprhdl) const override;
+}; // class CPhysicalAppendTableScan
+
+} // namespace gpopt
+
+#endif // !GPOPT_CPhysicalAppendTableScan_H
+
+// EOF
diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelAppendTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelAppendTableScan.h
new file mode 100644
index 00000000000..a7c4854507a
--- /dev/null
+++ b/src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelAppendTableScan.h
@@ -0,0 +1,143 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalParallelAppendTableScan.h
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/include/gpopt/operators/CPhysicalParallelAppendTableScan.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef GPOPT_CPhysicalParallelAppendTableScan_H
+#define GPOPT_CPhysicalParallelAppendTableScan_H
+
+#include "gpos/base.h"
+
+#include "gpopt/operators/CPhysicalDynamicScan.h"
+
+namespace gpopt
+{
+//---------------------------------------------------------------------------
+// @class:
+// CPhysicalParallelAppendTableScan
+//
+// @doc:
+// Parallel Append Table scan operator
+//
+//---------------------------------------------------------------------------
+class CPhysicalParallelAppendTableScan : public CPhysicalDynamicScan
+{
+private:
+ // number of parallel workers
+ ULONG m_ulParallelWorkers;
+
+ // worker-level distribution spec
+ CDistributionSpec *m_pdsWorkerDistribution;
+public:
+ CPhysicalParallelAppendTableScan(const CPhysicalParallelAppendTableScan &) = delete;
+
+ // ctors
+ CPhysicalParallelAppendTableScan(CMemoryPool *mp, const CName *pnameAlias,
+ CTableDescriptor *ptabdesc, ULONG ulOriginOpId,
+ ULONG scan_id, CColRefArray *pdrgpcrOutput,
+ CColRef2dArray *pdrgpdrgpcrParts,
+ IMdIdArray *partition_mdids,
+ ColRefToUlongMapArray *root_col_mapping_per_part,
+ ULONG ulParallelWorkers);
+
+ // dtor
+ ~CPhysicalParallelAppendTableScan() override;
+
+ // ident accessors
+ EOperatorId
+ Eopid() const override
+ {
+ return EopPhysicalParallelAppendTableScan;
+ }
+
+ // return a string for operator name
+ const CHAR *
+ SzId() const override
+ {
+ return "CPhysicalParallelAppendTableScan";
+ }
+
+ // match function
+ BOOL Matches(COperator *) const override;
+
+ // statistics derivation during costing
+ IStatistics *PstatsDerive(CMemoryPool *mp, CExpressionHandle &exprhdl,
+ CReqdPropPlan *prpplan,
+ IStatisticsArray *stats_ctxt) const override;
+
+ // number of parallel workers
+ ULONG UlParallelWorkers() const
+ {
+ return m_ulParallelWorkers;
+ }
+
+ // conversion function
+ static CPhysicalParallelAppendTableScan *
+ PopConvert(COperator *pop)
+ {
+ GPOS_ASSERT(nullptr != pop);
+ GPOS_ASSERT(EopPhysicalParallelAppendTableScan == pop->Eopid());
+
+ return dynamic_cast<CPhysicalParallelAppendTableScan *>(pop);
+ }
+
+ CPartitionPropagationSpec *PppsDerive(
+ CMemoryPool *mp, CExpressionHandle &exprhdl) const override;
+
+ 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 CPhysicalParallelAppendTableScan
+
+} // namespace gpopt
+
+#endif // !GPOPT_CPhysicalParallelAppendTableScan_H
+
+// EOF
diff --git a/src/backend/gporca/libgpopt/include/gpopt/translate/CTranslatorExprToDXL.h b/src/backend/gporca/libgpopt/include/gpopt/translate/CTranslatorExprToDXL.h
index 2c50e9e812f..db4b9f3532f 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/translate/CTranslatorExprToDXL.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/translate/CTranslatorExprToDXL.h
@@ -101,6 +101,9 @@ private:
// id of master node
INT m_iMasterId;
+ // whether is set operator
+ BOOL m_isSet = false;
+
// private copy ctor
CTranslatorExprToDXL(const CTranslatorExprToDXL &);
@@ -346,6 +349,32 @@ private:
CExpression *pexprScalarCond,
CDXLPhysicalProperties *dxl_properties);
+ // translate an append table scan
+ CDXLNode *PdxlnAppendTableScan(CExpression *pexprDTS,
+ CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ ULONG *pulNonGatherMotions, BOOL *pfDML);
+
+ // translate an append table scan with a scalar condition
+ CDXLNode *PdxlnAppendTableScan(CExpression *pexprDTS,
+ CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ CExpression *pexprScalarCond,
+ CDXLPhysicalProperties *dxl_properties);
+
+ // translate a parallel append table scan
+ CDXLNode *PdxlnParallelAppendTableScan(CExpression *pexprDTS,
+ CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ ULONG *pulNonGatherMotions, BOOL *pfDML);
+
+ // translate a parallel append table scan with a scalar condition
+ CDXLNode *PdxlnParallelAppendTableScan(CExpression *pexprDTS,
+ CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ CExpression *pexprScalarCond,
+ CDXLPhysicalProperties *dxl_properties);
+
// translate a dynamic bitmap table scan
CDXLNode *PdxlnDynamicBitmapTableScan(
CExpression *pexprDynamicBitmapTableScan, CColRefArray *colref_array,
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h
index d0fca8219ac..79d1652661c 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXform.h
@@ -72,6 +72,8 @@ public:
ExfGet2ParallelTableScan,
ExfIndexGet2IndexScan,
ExfDynamicGet2DynamicTableScan,
+ ExfDynamicGet2AppendTableScan,
+ ExfDynamicGet2ParallelAppendTableScan,
ExfDynamicIndexGet2DynamicIndexScan,
ExfImplementSequence,
ExfImplementConstTableGet,
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2AppendTableScan.h
similarity index 65%
copy from src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
copy to src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2AppendTableScan.h
index 99c8d4863d4..9bb4fca509a 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2AppendTableScan.h
@@ -17,18 +17,19 @@
* specific language governing permissions and limitations
* under the License.
*
- * CXformGet2ParallelTableScan.h
+ * CXformDynamicGet2AppendTableScan.h
*
* IDENTIFICATION
- * src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
+ * src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2AppendTableScan.h
*
*-------------------------------------------------------------------------
*/
-#ifndef GPOPT_CXformGet2ParallelTableScan_H
-#define GPOPT_CXformGet2ParallelTableScan_H
+#ifndef GPOPT_CXformDynamicGet2AppendTableScan_H
+#define GPOPT_CXformDynamicGet2AppendTableScan_H
#include "gpos/base.h"
+#include "gpopt/operators/CLogicalDynamicGet.h"
#include "gpopt/xforms/CXformImplementation.h"
namespace gpopt
@@ -37,39 +38,37 @@ using namespace gpos;
//---------------------------------------------------------------------------
// @class:
-// CXformGet2ParallelTableScan
+// CXformDynamicGet2Append
//
// @doc:
-// Transform Get to Parallel TableScan using GUC enable_parallel
+// Transform DynamicGet to Append Table Scan
//
//---------------------------------------------------------------------------
-class CXformGet2ParallelTableScan : public CXformImplementation
+class CXformDynamicGet2AppendTableScan : 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;
+ CXformDynamicGet2AppendTableScan(
+ const CXformDynamicGet2AppendTableScan &) = delete;
// ctor
- explicit CXformGet2ParallelTableScan(CMemoryPool *);
+ explicit CXformDynamicGet2AppendTableScan(CMemoryPool *mp);
// dtor
- ~CXformGet2ParallelTableScan() override = default;
+ ~CXformDynamicGet2AppendTableScan() override = default;
// ident accessors
EXformId
Exfid() const override
{
- return ExfGet2ParallelTableScan;
+ return ExfDynamicGet2AppendTableScan;
}
// return a string for xform name
const CHAR *
SzId() const override
{
- return "CXformGet2ParallelTableScan";
+ return "CXformDynamicGet2AppendTableScan";
}
// compute xform promise for a given expression handle
@@ -79,10 +78,10 @@ public:
void Transform(CXformContext *pxfctxt, CXformResult *pxfres,
CExpression *pexpr) const override;
-}; // class CXformGet2ParallelTableScan
+}; // class CXformDynamicGet2AppendTableScan
-} // namespace gpopt
+} // namespace gpopt
-#endif // !GPOPT_CXformGet2ParallelTableScan_H
+#endif // !GPOPT_CXformDynamicGet2AppendTableScan_H
-// EOF
\ No newline at end of file
+// EOF
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h
similarity index 64%
copy from src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
copy to src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h
index 99c8d4863d4..7036d7fb767 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h
@@ -17,18 +17,19 @@
* specific language governing permissions and limitations
* under the License.
*
- * CXformGet2ParallelTableScan.h
+ * CXformDynamicGet2ParallelAppendTableScan.h
*
* IDENTIFICATION
- * src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
+ * src/backend/gporca/libgpopt/include/gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h
*
*-------------------------------------------------------------------------
*/
-#ifndef GPOPT_CXformGet2ParallelTableScan_H
-#define GPOPT_CXformGet2ParallelTableScan_H
+#ifndef GPOPT_CXformDynamicGet2ParallelAppendTableScan_H
+#define GPOPT_CXformDynamicGet2ParallelAppendTableScan_H
#include "gpos/base.h"
+#include "gpopt/operators/CLogicalDynamicGet.h"
#include "gpopt/xforms/CXformImplementation.h"
namespace gpopt
@@ -37,39 +38,37 @@ using namespace gpos;
//---------------------------------------------------------------------------
// @class:
-// CXformGet2ParallelTableScan
+// CXformDynamicGet2ParallelAppendTableScan
//
// @doc:
-// Transform Get to Parallel TableScan using GUC enable_parallel
+// Transform DynamicGet to Parallel Append Table Scan
//
//---------------------------------------------------------------------------
-class CXformGet2ParallelTableScan : public CXformImplementation
+class CXformDynamicGet2ParallelAppendTableScan : 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;
+ CXformDynamicGet2ParallelAppendTableScan(
+ const CXformDynamicGet2ParallelAppendTableScan &) = delete;
// ctor
- explicit CXformGet2ParallelTableScan(CMemoryPool *);
+ explicit CXformDynamicGet2ParallelAppendTableScan(CMemoryPool *mp);
// dtor
- ~CXformGet2ParallelTableScan() override = default;
+ ~CXformDynamicGet2ParallelAppendTableScan() override = default;
// ident accessors
EXformId
Exfid() const override
{
- return ExfGet2ParallelTableScan;
+ return ExfDynamicGet2ParallelAppendTableScan;
}
// return a string for xform name
const CHAR *
SzId() const override
{
- return "CXformGet2ParallelTableScan";
+ return "CXformDynamicGet2ParallelAppendTableScan";
}
// compute xform promise for a given expression handle
@@ -79,10 +78,10 @@ public:
void Transform(CXformContext *pxfctxt, CXformResult *pxfres,
CExpression *pexpr) const override;
-}; // class CXformGet2ParallelTableScan
+}; // class CXformDynamicGet2ParallelAppendTableScan
-} // namespace gpopt
+} // namespace gpopt
-#endif // !GPOPT_CXformGet2ParallelTableScan_H
+#endif // !GPOPT_CXformDynamicGet2ParallelAppendTableScan_H
-// EOF
\ No newline at end of file
+// EOF
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
index 99c8d4863d4..ce43b241cd1 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CXformGet2ParallelTableScan.h
@@ -45,11 +45,11 @@ using namespace gpos;
//---------------------------------------------------------------------------
class CXformGet2ParallelTableScan : public CXformImplementation
{
-private:
+public:
// check if memo contains logical operators that are incompatible with parallel execution
+ // (shared with other parallel scan xforms, e.g. CXformDynamicGet2ParallelAppendTableScan)
static BOOL FHasParallelIncompatibleOps(CExpressionHandle &exprhdl);
-public:
CXformGet2ParallelTableScan(const CXformGet2ParallelTableScan &) = delete;
// ctor
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h
index 0ee9beb2eb7..2d788d6c6b4 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/xforms.h
@@ -30,6 +30,8 @@
#include "gpopt/xforms/CXformDifferenceAll2LeftAntiSemiJoin.h"
#include "gpopt/xforms/CXformDynamicForeignGet2DynamicForeignScan.h"
#include "gpopt/xforms/CXformDynamicGet2DynamicTableScan.h"
+#include "gpopt/xforms/CXformDynamicGet2AppendTableScan.h"
+#include "gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h"
#include "gpopt/xforms/CXformDynamicIndexGet2DynamicIndexScan.h"
#include "gpopt/xforms/CXformDynamicIndexOnlyGet2DynamicIndexOnlyScan.h"
#include "gpopt/xforms/CXformEagerAgg.h"
diff --git a/src/backend/gporca/libgpopt/src/base/CQueryContext.cpp b/src/backend/gporca/libgpopt/src/base/CQueryContext.cpp
index 4c75e7202ed..b3715fc599e 100644
--- a/src/backend/gporca/libgpopt/src/base/CQueryContext.cpp
+++ b/src/backend/gporca/libgpopt/src/base/CQueryContext.cpp
@@ -240,8 +240,8 @@ CQueryContext::PqcGenerate(CMemoryPool *mp, CExpression *pexpr,
GPOS_NEW(mp) CEnfdRewindability(prs, CEnfdRewindability::ErmSatisfy);
CEnfdPartitionPropagation *pepp = GPOS_NEW(mp)
CEnfdPartitionPropagation(ppps, CEnfdPartitionPropagation::EppmSatisfy);
-
- // Required CTEs are obtained from the CTEInfo global information in the optimizer context
+
+ // Required CTEs are obtained from the CTEInfo global information in the optimizer context
CCTEReq *pcter = poptctxt->Pcteinfo()->PcterProducers(mp);
// NB: Partition propagation requirements are not initialized here. They are
diff --git a/src/backend/gporca/libgpopt/src/operators/CLogicalDynamicGet.cpp b/src/backend/gporca/libgpopt/src/operators/CLogicalDynamicGet.cpp
index 3fbefec189d..2b6adc43a8f 100644
--- a/src/backend/gporca/libgpopt/src/operators/CLogicalDynamicGet.cpp
+++ b/src/backend/gporca/libgpopt/src/operators/CLogicalDynamicGet.cpp
@@ -244,6 +244,8 @@ CLogicalDynamicGet::PxfsCandidates(CMemoryPool *mp) const
(void) xform_set->ExchangeSet(CXform::ExfDynamicGet2DynamicTableScan);
(void) xform_set->ExchangeSet(
CXform::ExfExpandDynamicGetWithForeignPartitions);
+ (void) xform_set->ExchangeSet(CXform::ExfDynamicGet2AppendTableScan);
+ (void) xform_set->ExchangeSet(CXform::ExfDynamicGet2ParallelAppendTableScan);
return xform_set;
}
diff --git a/src/backend/gporca/libgpopt/src/operators/CPhysicalAppendTableScan.cpp b/src/backend/gporca/libgpopt/src/operators/CPhysicalAppendTableScan.cpp
new file mode 100644
index 00000000000..66592599609
--- /dev/null
+++ b/src/backend/gporca/libgpopt/src/operators/CPhysicalAppendTableScan.cpp
@@ -0,0 +1,111 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalAppendTableScan.cpp
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/src/operators/CPhysicalAppendTableScan.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "gpopt/operators/CPhysicalAppendTableScan.h"
+
+#include "gpos/base.h"
+
+#include "gpopt/base/CDistributionSpec.h"
+#include "gpopt/base/CDistributionSpecHashed.h"
+#include "gpopt/base/CDistributionSpecRandom.h"
+#include "gpopt/base/CDistributionSpecSingleton.h"
+#include "gpopt/base/CUtils.h"
+#include "gpopt/metadata/CName.h"
+#include "gpopt/metadata/CTableDescriptor.h"
+#include "naucrates/statistics/CStatisticsUtils.h"
+
+using namespace gpopt;
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalAppendTableScan::CPhysicalAppendTableScan
+//
+// @doc:
+// Ctor
+//
+//---------------------------------------------------------------------------
+CPhysicalAppendTableScan::CPhysicalAppendTableScan(CMemoryPool *mp,
+ const CName *pnameAlias,
+ CTableDescriptor *ptabdesc,
+ ULONG ulOriginOpId,
+ ULONG scan_id,
+ CColRefArray *pdrgpcrOutput,
+ CColRef2dArray *pdrgpdrgpcrParts,
+ IMdIdArray *partition_mdids,
+ ColRefToUlongMapArray *root_col_mapping_per_part)
+ : CPhysicalDynamicScan(mp, ptabdesc, ulOriginOpId, pnameAlias, scan_id,
+ pdrgpcrOutput, pdrgpdrgpcrParts, partition_mdids,
+ root_col_mapping_per_part)
+{
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalAppendTableScan::Matches
+//
+// @doc:
+// match operator
+//
+//---------------------------------------------------------------------------
+BOOL
+CPhysicalAppendTableScan::Matches(COperator *pop) const
+{
+ return CUtils::FMatchDynamicScan(this, pop);
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalAppendTableScan::PstatsDerive
+//
+// @doc:
+// Statistics derivation during costing
+//
+//---------------------------------------------------------------------------
+IStatistics *
+CPhysicalAppendTableScan::PstatsDerive(CMemoryPool *mp,
+ CExpressionHandle &exprhdl,
+ CReqdPropPlan *prpplan,
+ IStatisticsArray * // stats_ctxt
+) const
+{
+ GPOS_ASSERT(nullptr != prpplan);
+
+ return CStatisticsUtils::DeriveStatsForDynamicScan(
+ mp, exprhdl, ScanId(), prpplan->Pepp()->PppsRequired());
+}
+
+CPartitionPropagationSpec *
+CPhysicalAppendTableScan::PppsDerive(CMemoryPool *mp, CExpressionHandle &) const
+{
+ CPartitionPropagationSpec *pps = GPOS_NEW(mp) CPartitionPropagationSpec(mp);
+ pps->Insert(ScanId(), CPartitionPropagationSpec::EpptConsumer,
+ Ptabdesc()->MDId(), nullptr, nullptr);
+
+ return pps;
+}
+
+// EOF
diff --git a/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelAppendTableScan.cpp b/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelAppendTableScan.cpp
new file mode 100644
index 00000000000..565c42ceeb8
--- /dev/null
+++ b/src/backend/gporca/libgpopt/src/operators/CPhysicalParallelAppendTableScan.cpp
@@ -0,0 +1,271 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalAppendTableScan.cpp
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/src/operators/CPhysicalAppendTableScan.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "gpopt/operators/CPhysicalAppendTableScan.h"
+#include "gpopt/operators/CPhysicalParallelAppendTableScan.h"
+
+#include "gpos/base.h"
+
+#include "gpopt/base/CDistributionSpec.h"
+#include "gpopt/base/CDistributionSpecHashed.h"
+#include "gpopt/base/CDistributionSpecRandom.h"
+#include "gpopt/base/CDistributionSpecSingleton.h"
+#include "gpopt/base/CDistributionSpecWorkerRandom.h"
+#include "gpopt/base/CUtils.h"
+#include "gpopt/metadata/CName.h"
+#include "gpopt/metadata/CTableDescriptor.h"
+#include "naucrates/statistics/CStatisticsUtils.h"
+
+using namespace gpopt;
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalParallelAppendTableScan::CPhysicalParallelAppendTableScan
+//
+// @doc:
+// Ctor
+//
+//---------------------------------------------------------------------------
+CPhysicalParallelAppendTableScan::CPhysicalParallelAppendTableScan(CMemoryPool *mp,
+ const CName *pnameAlias,
+ CTableDescriptor *ptabdesc,
+ ULONG ulOriginOpId,
+ ULONG scan_id,
+ CColRefArray *pdrgpcrOutput,
+ CColRef2dArray *pdrgpdrgpcrParts,
+ IMdIdArray *partition_mdids,
+ ColRefToUlongMapArray *root_col_mapping_per_part,
+ ULONG ulParallelWorkers)
+ : CPhysicalDynamicScan(mp, ptabdesc, ulOriginOpId, pnameAlias, scan_id,
+ pdrgpcrOutput, pdrgpdrgpcrParts, partition_mdids,
+ root_col_mapping_per_part),
+ 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:
+// CPhysicalParallelAppendTableScan::~CPhysicalParallelAppendTableScan
+//
+// @doc:
+// dtor
+//
+//---------------------------------------------------------------------------
+CPhysicalParallelAppendTableScan::~CPhysicalParallelAppendTableScan()
+{
+ CRefCount::SafeRelease(m_pdsWorkerDistribution);
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalParallelAppendTableScan::Matches
+//
+// @doc:
+// match operator
+//
+//---------------------------------------------------------------------------
+BOOL
+CPhysicalParallelAppendTableScan::Matches(COperator *pop) const
+{
+ if (Eopid() != pop->Eopid())
+ {
+ return false;
+ }
+
+ CPhysicalParallelAppendTableScan *popParallelAppendTableScan =
+ CPhysicalParallelAppendTableScan::PopConvert(pop);
+
+ return m_ptabdesc->MDId()->Equals(popParallelAppendTableScan->Ptabdesc()->MDId()) &&
+ m_pdrgpcrOutput->Equals(popParallelAppendTableScan->PdrgpcrOutput()) &&
+ m_ulParallelWorkers == popParallelAppendTableScan->UlParallelWorkers();
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalParallelAppendTableScan::PstatsDerive
+//
+// @doc:
+// Statistics derivation during costing
+//
+//---------------------------------------------------------------------------
+IStatistics *
+CPhysicalParallelAppendTableScan::PstatsDerive(CMemoryPool *mp,
+ CExpressionHandle &exprhdl,
+ CReqdPropPlan *prpplan,
+ IStatisticsArray * // stats_ctxt
+) const
+{
+ GPOS_ASSERT(nullptr != prpplan);
+
+ return CStatisticsUtils::DeriveStatsForDynamicScan(
+ mp, exprhdl, ScanId(), prpplan->Pepp()->PppsRequired());
+}
+
+CPartitionPropagationSpec *
+CPhysicalParallelAppendTableScan::PppsDerive(CMemoryPool *mp, CExpressionHandle &) const
+{
+ CPartitionPropagationSpec *pps = GPOS_NEW(mp) CPartitionPropagationSpec(mp);
+ pps->Insert(ScanId(), CPartitionPropagationSpec::EpptConsumer,
+ Ptabdesc()->MDId(), nullptr, nullptr);
+
+ return pps;
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CPhysicalParallelAppendTableScan::PdsDerive
+//
+// @doc:
+// Derive distribution for parallel append table scan
+//
+//---------------------------------------------------------------------------
+CDistributionSpec *
+CPhysicalParallelAppendTableScan::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:
+// CPhysicalParallelAppendTableScan::EpetDistribution
+//
+// @doc:
+// Return the enforcing type for distribution property based on this
+// operator
+//
+//---------------------------------------------------------------------------
+CEnfdProp::EPropEnforcingType
+CPhysicalParallelAppendTableScan::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:
+// CPhysicalParallelAppendTableScan::EpetRewindability
+//
+// @doc:
+// Return rewindability property enforcing type for this operator
+//
+//---------------------------------------------------------------------------
+CEnfdProp::EPropEnforcingType
+CPhysicalParallelAppendTableScan::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:
+// CPhysicalParallelAppendTableScan::FValidContext
+//
+// @doc:
+// Check if optimization contexts is valid;
+// Reject if parent requires REWINDABLE (e.g., for NL Join inner child)
+// because ParallelAppendTableScan derives NONE (not rewindable)
+//
+//---------------------------------------------------------------------------
+BOOL
+CPhysicalParallelAppendTableScan::FValidContext(CMemoryPool *,
+ COptimizationContext *poc,
+ COptimizationContextArray *pdrgpocChild) 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;
+ }
+
+ if (nullptr != pdrgpocChild && pdrgpocChild->Size() == 1)
+ {
+ CEnfdDistribution *ed = (*pdrgpocChild)[0]->Prpp()->Ped();
+ CDistributionSpec *dsReq = ed->PdsRequired();
+
+ const CDistributionSpec::EDistributionType dt = dsReq->Edt();
+ if (dt != CDistributionSpec::EdtWorkerRandom ||
+ dt != CDistributionSpec::EdtReplicatedWorkers)
+ return false;
+ }
+
+ return true;
+}
+
+// EOF
diff --git a/src/backend/gporca/libgpopt/src/operators/Makefile b/src/backend/gporca/libgpopt/src/operators/Makefile
index ce5362e9049..f9b0fbec73c 100644
--- a/src/backend/gporca/libgpopt/src/operators/Makefile
+++ b/src/backend/gporca/libgpopt/src/operators/Makefile
@@ -93,6 +93,8 @@ OBJS = CExpression.o \
CPhysicalDynamicIndexScan.o \
CPhysicalDynamicScan.o \
CPhysicalDynamicTableScan.o \
+ CPhysicalAppendTableScan.o \
+ CPhysicalParallelAppendTableScan.o \
CPhysicalForeignScan.o \
CPhysicalFilter.o \
CPhysicalFullMergeJoin.o \
diff --git a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp
index 51acac518be..933d606a7d8 100644
--- a/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp
+++ b/src/backend/gporca/libgpopt/src/translate/CTranslatorExprToDXL.cpp
@@ -30,6 +30,8 @@
#include "gpopt/exception.h"
#include "gpopt/mdcache/CMDAccessorUtils.h"
#include "gpopt/operators/CPhysicalAgg.h"
+#include "gpopt/operators/CPhysicalAppendTableScan.h"
+#include "gpopt/operators/CPhysicalParallelAppendTableScan.h"
#include "gpopt/operators/CPhysicalAssert.h"
#include "gpopt/operators/CPhysicalBitmapTableScan.h"
#include "gpopt/operators/CPhysicalCTEConsumer.h"
@@ -100,6 +102,7 @@
#include "naucrates/dxl/operators/CDXLDatumBool.h"
#include "naucrates/dxl/operators/CDXLDirectDispatchInfo.h"
#include "naucrates/dxl/operators/CDXLPhysicalAppend.h"
+#include "naucrates/dxl/operators/CDXLPhysicalParallelAppend.h"
#include "naucrates/dxl/operators/CDXLPhysicalAssert.h"
#include "naucrates/dxl/operators/CDXLPhysicalBitmapTableScan.h"
#include "naucrates/dxl/operators/CDXLPhysicalBroadcastMotion.h"
@@ -369,6 +372,13 @@ CTranslatorExprToDXL::CreateDXLNode(CExpression *pexpr,
return dxlnode;
}
+ else if (COperator::EopPhysicalParallelAppendTableScan == ulOpId && m_isSet)
+ {
+ CDXLNode *dxlnode = PdxlnAppendTableScan(
+ pexpr, colref_array, pdrgpdsBaseTables, pulNonGatherMotions, pfDML);
+
+ return dxlnode;
+ }
// add a result node on top to project out columns not needed any further,
// for instance, if the grouping /order by /partition/ distribution columns
// are no longer needed
@@ -487,6 +497,16 @@ CTranslatorExprToDXL::CreateDXLNode(CExpression *pexpr,
pexpr, colref_array, pdrgpdsBaseTables, pulNonGatherMotions,
pfDML);
break;
+ case COperator::EopPhysicalAppendTableScan:
+ dxlnode = CTranslatorExprToDXL::PdxlnAppendTableScan(
+ pexpr, colref_array, pdrgpdsBaseTables, pulNonGatherMotions,
+ pfDML);
+ break;
+ case COperator::EopPhysicalParallelAppendTableScan:
+ dxlnode = CTranslatorExprToDXL::PdxlnParallelAppendTableScan(
+ pexpr, colref_array, pdrgpdsBaseTables, pulNonGatherMotions,
+ pfDML);
+ break;
case COperator::EopPhysicalDynamicBitmapTableScan:
dxlnode = CTranslatorExprToDXL::PdxlnDynamicBitmapTableScan(
pexpr, colref_array, pdrgpdsBaseTables, pulNonGatherMotions,
@@ -1288,6 +1308,52 @@ CTranslatorExprToDXL::PdxlnDynamicTableScan(
pexprScalarCond, dxl_properties);
}
+//---------------------------------------------------------------------------
+// @function:
+// CTranslatorExprToDXL::PdxlnAppendTableScan
+//
+// @doc:
+// Create a DXL append table scan node from an optimizer
+// append table scan node.
+//
+//---------------------------------------------------------------------------
+CDXLNode *
+CTranslatorExprToDXL::PdxlnAppendTableScan(
+ CExpression *pexprDTS, CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ ULONG *, // pulNonGatherMotions,
+ BOOL * // pfDML
+)
+{
+ CExpression *pexprScalarCond = nullptr;
+ CDXLPhysicalProperties *dxl_properties = nullptr;
+ return PdxlnAppendTableScan(pexprDTS, colref_array, pdrgpdsBaseTables,
+ pexprScalarCond, dxl_properties);
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CTranslatorExprToDXL::PdxlnParallelAppendTableScan
+//
+// @doc:
+// Create a DXL parallel append table scan node from an optimizer
+// parallel append table scan node.
+//
+//---------------------------------------------------------------------------
+CDXLNode *
+CTranslatorExprToDXL::PdxlnParallelAppendTableScan(
+ CExpression *pexprDTS, CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables,
+ ULONG *, // pulNonGatherMotions,
+ BOOL * // pfDML
+)
+{
+ CExpression *pexprScalarCond = nullptr;
+ CDXLPhysicalProperties *dxl_properties = nullptr;
+ return PdxlnParallelAppendTableScan(pexprDTS, colref_array, pdrgpdsBaseTables,
+ pexprScalarCond, dxl_properties);
+}
+
// Construct a dxl table descr for a child partition
CTableDescriptor *
CTranslatorExprToDXL::MakeTableDescForPart(const IMDRelation *part,
@@ -1458,6 +1524,291 @@ CTranslatorExprToDXL::PdxlnDynamicTableScan(
return pdxlnDTS;
}
+// Translate CPhysicalAppendTableScan node. It creates a CDXLPhysicalAppend
+// node over a number of CDXLPhysicalTableScan nodes - include all child partition
+// tables of the root partition.
+//
+// To handle dropped and re-ordered columns, the project list and any filter
+// expression from the root table are modified using the per partition mappings
+// for each child CDXLPhysicalTableScan
+CDXLNode *
+CTranslatorExprToDXL::PdxlnAppendTableScan(
+ CExpression *pexprDTS, CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables, CExpression *pexprScalarCond,
+ CDXLPhysicalProperties *dxl_properties)
+{
+ CPhysicalDynamicScan *popDTS =
+ CPhysicalDynamicScan::PopConvert(pexprDTS->Pop());
+
+ ULongPtrArray *selector_ids = GPOS_NEW(m_mp) ULongPtrArray(m_mp);
+ CPartitionPropagationSpec *pps_reqd =
+ pexprDTS->Prpp()->Pepp()->PppsRequired();
+ if (pps_reqd->Contains(popDTS->ScanId()))
+ {
+ const CBitSet *bs = pps_reqd->SelectorIds(popDTS->ScanId());
+ CBitSetIter bsi(*bs);
+ for (ULONG ul = 0; bsi.Advance(); ul++)
+ {
+ selector_ids->Append(GPOS_NEW(m_mp) ULONG(bsi.Bit()));
+ }
+ }
+
+ // construct plan costs
+ CDXLPhysicalProperties *pdxlpropDTS = GetProperties(pexprDTS);
+
+ if (nullptr != dxl_properties)
+ {
+ CWStringDynamic *rows_out_str = GPOS_NEW(m_mp) CWStringDynamic(
+ m_mp,
+ dxl_properties->GetDXLOperatorCost()->GetRowsOutStr()->GetBuffer());
+ CWStringDynamic *pstrCost = GPOS_NEW(m_mp)
+ CWStringDynamic(m_mp, dxl_properties->GetDXLOperatorCost()
+ ->GetTotalCostStr()
+ ->GetBuffer());
+
+ pdxlpropDTS->GetDXLOperatorCost()->SetRows(rows_out_str);
+ pdxlpropDTS->GetDXLOperatorCost()->SetCost(pstrCost);
+ dxl_properties->Release();
+ }
+ GPOS_ASSERT(nullptr != pexprDTS->Prpp());
+
+ // construct projection list for top-level Append node
+ CColRefSet *pcrsOutput = pexprDTS->Prpp()->PcrsRequired();
+ CDXLNode *pdxlnPrLAppend = PdxlnProjList(pcrsOutput, colref_array);
+ CDXLTableDescr *root_dxl_table_descr = MakeDXLTableDescr(
+ popDTS->Ptabdesc(), popDTS->PdrgpcrOutput(), pexprDTS->Prpp());
+
+ // Construct the Append node - even when there is only one child partition.
+ // This is done for two reasons:
+ // * Dynamic partition pruning
+ // Even if one partition is present in the statically pruned plan, we could
+ // still dynamically prune it away. This needs an Append node.
+ // * Col mappings issues
+ // When the first selected child partition's cols have different types/order
+ // than the root partition, we can no longer re-use the colrefs of the root
+ // partition, since colrefs are immutable. Thus, we create new colrefs for
+ // this partition. But, if there is no Append (in case of just one selected
+ // partition), then we also go through update all references above the DTS
+ // with the new colrefs. For simplicity, we decided to keep the Append
+ // around to maintain this projection (mapping) from the old root colrefs
+ // to the first selected partition colrefs.
+ //
+ // GPDB_12_MERGE_FIXME: An Append on a single TableScan can be removed in
+ // CTranslatorDXLToPlstmt since these points do not apply there.
+ CDXLNode *pdxlnAppend = GPOS_NEW(m_mp) CDXLNode(
+ m_mp,
+ GPOS_NEW(m_mp) CDXLPhysicalAppend(m_mp, false, false, popDTS->ScanId(),
+ root_dxl_table_descr, selector_ids));
+ pdxlnAppend->SetProperties(pdxlpropDTS);
+ pdxlnAppend->AddChild(pdxlnPrLAppend);
+ pdxlnAppend->AddChild(PdxlnFilter(nullptr));
+
+ IMdIdArray *part_mdids = popDTS->GetPartitionMdids();
+ for (ULONG ul = 0; ul < part_mdids->Size(); ++ul)
+ {
+ IMDId *part_mdid = (*part_mdids)[ul];
+ const IMDRelation *part = m_pmda->RetrieveRel(part_mdid);
+
+ CTableDescriptor *part_tabdesc =
+ MakeTableDescForPart(part, popDTS->Ptabdesc());
+
+ // Create new colrefs for the child partition. The ColRefs from root
+ // DTS, which may be used in any parent node, can no longer be exported
+ // by a child of the Append node. Thus it is exported by the Append
+ // node itself, and new colrefs are created here.
+ CColRefArray *part_colrefs = GPOS_NEW(m_mp) CColRefArray(m_mp);
+ for (ULONG ul_col = 0; ul_col < part_tabdesc->ColumnCount(); ++ul_col)
+ {
+ const CColumnDescriptor *cd = part_tabdesc->Pcoldesc(ul_col);
+ CColRef *cr = m_pcf->PcrCreate(cd->RetrieveType(),
+ cd->TypeModifier(), cd->Name());
+ part_colrefs->Append(cr);
+ }
+
+ CDXLTableDescr *dxl_table_descr =
+ MakeDXLTableDescr(part_tabdesc, part_colrefs, pexprDTS->Prpp());
+ part_tabdesc->Release();
+
+ CDXLNode *dxlnode = GPOS_NEW(m_mp) CDXLNode(
+ m_mp, GPOS_NEW(m_mp) CDXLPhysicalTableScan(m_mp, dxl_table_descr));
+
+ // GPDB_12_MERGE_FIXME: Compute stats & properties per scan
+ pdxlpropDTS->AddRef();
+ dxlnode->SetProperties(pdxlpropDTS);
+
+ // ColRef -> index in child table desc (per partition)
+ auto root_col_mapping = (*popDTS->GetRootColMappingPerPart())[ul];
+
+ // construct projection list, re-ordered to match root DTS
+ CDXLNode *pdxlnPrL = PdxlnProjListForChildPart(
+ root_col_mapping, part_colrefs, pcrsOutput, colref_array);
+ dxlnode->AddChild(pdxlnPrL); // project list
+
+ // construct the filter
+ CDXLNode *filter_dxlnode = PdxlnFilter(
+ PdxlnCondForChildPart(root_col_mapping, part_colrefs,
+ popDTS->PdrgpcrOutput(), pexprScalarCond));
+ dxlnode->AddChild(filter_dxlnode); // filter
+
+ // add to the other scans under the created Append node
+ pdxlnAppend->AddChild(dxlnode);
+
+ // cleanup
+ part_colrefs->Release();
+ }
+
+ CDistributionSpec *pds = pexprDTS->GetDrvdPropPlan()->Pds();
+ pds->AddRef();
+ pdrgpdsBaseTables->Append(pds);
+
+ GPOS_ASSERT(pdxlnAppend);
+ return pdxlnAppend;
+}
+
+// Translate CPhysicalParallelAppendTableScan node. It creates a CDXLPhysicalParallelAppend
+// node over a number of CDXLPhysicalParallelTableScan nodes - include all child partition
+// tables of the root partition.
+//
+// To handle dropped and re-ordered columns, the project list and any filter
+// expression from the root table are modified using the per partition mappings
+// for each child CDXLPhysicalParallelTableScan
+CDXLNode *
+CTranslatorExprToDXL::PdxlnParallelAppendTableScan(
+ CExpression *pexprDTS, CColRefArray *colref_array,
+ CDistributionSpecArray *pdrgpdsBaseTables, CExpression *pexprScalarCond,
+ CDXLPhysicalProperties *dxl_properties)
+{
+ ULONG ulParallelWorkers = 0;
+
+ CPhysicalParallelAppendTableScan *popDTS =
+ CPhysicalParallelAppendTableScan::PopConvert(pexprDTS->Pop());
+
+ ulParallelWorkers = popDTS->UlParallelWorkers();
+ ULongPtrArray *selector_ids = GPOS_NEW(m_mp) ULongPtrArray(m_mp);
+ CPartitionPropagationSpec *pps_reqd =
+ pexprDTS->Prpp()->Pepp()->PppsRequired();
+ if (pps_reqd->Contains(popDTS->ScanId()))
+ {
+ const CBitSet *bs = pps_reqd->SelectorIds(popDTS->ScanId());
+ CBitSetIter bsi(*bs);
+ for (ULONG ul = 0; bsi.Advance(); ul++)
+ {
+ selector_ids->Append(GPOS_NEW(m_mp) ULONG(bsi.Bit()));
+ }
+ }
+
+ // construct plan costs
+ CDXLPhysicalProperties *pdxlpropDTS = GetProperties(pexprDTS);
+
+ if (nullptr != dxl_properties)
+ {
+ CWStringDynamic *rows_out_str = GPOS_NEW(m_mp) CWStringDynamic(
+ m_mp,
+ dxl_properties->GetDXLOperatorCost()->GetRowsOutStr()->GetBuffer());
+ CWStringDynamic *pstrCost = GPOS_NEW(m_mp)
+ CWStringDynamic(m_mp, dxl_properties->GetDXLOperatorCost()
+ ->GetTotalCostStr()
+ ->GetBuffer());
+
+ pdxlpropDTS->GetDXLOperatorCost()->SetRows(rows_out_str);
+ pdxlpropDTS->GetDXLOperatorCost()->SetCost(pstrCost);
+ dxl_properties->Release();
+ }
+ GPOS_ASSERT(nullptr != pexprDTS->Prpp());
+
+ // construct projection list for top-level Append node
+ CColRefSet *pcrsOutput = pexprDTS->Prpp()->PcrsRequired();
+ CDXLNode *pdxlnPrLAppend = PdxlnProjList(pcrsOutput, colref_array);
+ CDXLTableDescr *root_dxl_table_descr = MakeDXLTableDescr(
+ popDTS->Ptabdesc(), popDTS->PdrgpcrOutput(), pexprDTS->Prpp());
+
+ // Construct the Append node - even when there is only one child partition.
+ // This is done for two reasons:
+ // * Dynamic partition pruning
+ // Even if one partition is present in the statically pruned plan, we could
+ // still dynamically prune it away. This needs an Append node.
+ // * Col mappings issues
+ // When the first selected child partition's cols have different types/order
+ // than the root partition, we can no longer re-use the colrefs of the root
+ // partition, since colrefs are immutable. Thus, we create new colrefs for
+ // this partition. But, if there is no Append (in case of just one selected
+ // partition), then we also go through update all references above the DTS
+ // with the new colrefs. For simplicity, we decided to keep the Append
+ // around to maintain this projection (mapping) from the old root colrefs
+ // to the first selected partition colrefs.
+ //
+ // GPDB_12_MERGE_FIXME: An Append on a single TableScan can be removed in
+ // CTranslatorDXLToPlstmt since these points do not apply there.
+ CDXLNode *pdxlnAppend = GPOS_NEW(m_mp) CDXLNode(
+ m_mp,
+ GPOS_NEW(m_mp) CDXLPhysicalParallelAppend(m_mp, false, false, popDTS->ScanId(),
+ root_dxl_table_descr, selector_ids, ulParallelWorkers));
+ pdxlnAppend->SetProperties(pdxlpropDTS);
+ pdxlnAppend->AddChild(pdxlnPrLAppend);
+ pdxlnAppend->AddChild(PdxlnFilter(nullptr));
+
+ IMdIdArray *part_mdids = popDTS->GetPartitionMdids();
+ for (ULONG ul = 0; ul < part_mdids->Size(); ++ul)
+ {
+ IMDId *part_mdid = (*part_mdids)[ul];
+ const IMDRelation *part = m_pmda->RetrieveRel(part_mdid);
+
+ CTableDescriptor *part_tabdesc =
+ MakeTableDescForPart(part, popDTS->Ptabdesc());
+
+ // Create new colrefs for the child partition. The ColRefs from root
+ // DTS, which may be used in any parent node, can no longer be exported
+ // by a child of the Append node. Thus it is exported by the Append
+ // node itself, and new colrefs are created here.
+ CColRefArray *part_colrefs = GPOS_NEW(m_mp) CColRefArray(m_mp);
+ for (ULONG ul_col = 0; ul_col < part_tabdesc->ColumnCount(); ++ul_col)
+ {
+ const CColumnDescriptor *cd = part_tabdesc->Pcoldesc(ul_col);
+ CColRef *cr = m_pcf->PcrCreate(cd->RetrieveType(),
+ cd->TypeModifier(), cd->Name());
+ part_colrefs->Append(cr);
+ }
+
+ CDXLTableDescr *dxl_table_descr =
+ MakeDXLTableDescr(part_tabdesc, part_colrefs, pexprDTS->Prpp());
+ part_tabdesc->Release();
+
+ CDXLNode *dxlnode = GPOS_NEW(m_mp) CDXLNode(
+ m_mp, GPOS_NEW(m_mp) CDXLPhysicalParallelTableScan(m_mp, dxl_table_descr, ulParallelWorkers));
+
+ // GPDB_12_MERGE_FIXME: Compute stats & properties per scan
+ pdxlpropDTS->AddRef();
+ dxlnode->SetProperties(pdxlpropDTS);
+
+ // ColRef -> index in child table desc (per partition)
+ auto root_col_mapping = (*popDTS->GetRootColMappingPerPart())[ul];
+
+ // construct projection list, re-ordered to match root DTS
+ CDXLNode *pdxlnPrL = PdxlnProjListForChildPart(
+ root_col_mapping, part_colrefs, pcrsOutput, colref_array);
+ dxlnode->AddChild(pdxlnPrL); // project list
+
+ // construct the filter
+ CDXLNode *filter_dxlnode = PdxlnFilter(
+ PdxlnCondForChildPart(root_col_mapping, part_colrefs,
+ popDTS->PdrgpcrOutput(), pexprScalarCond));
+ dxlnode->AddChild(filter_dxlnode); // filter
+
+ // add to the other scans under the created Append node
+ pdxlnAppend->AddChild(dxlnode);
+
+ // cleanup
+ part_colrefs->Release();
+ }
+
+ CDistributionSpec *pds = pexprDTS->GetDrvdPropPlan()->Pds();
+ pds->AddRef();
+ pdrgpdsBaseTables->Append(pds);
+
+ GPOS_ASSERT(pdxlnAppend);
+ return pdxlnAppend;
+}
+
//---------------------------------------------------------------------------
// @function:
// CTranslatorExprToDXL::PdxlnDynamicScanToAppend
@@ -2638,6 +2989,20 @@ CTranslatorExprToDXL::PdxlnFromFilter(CExpression *pexprFilter,
pdrgpdsBaseTables, pexprScalar,
dxl_properties);
}
+ case COperator::EopPhysicalAppendTableScan:
+ {
+ dxl_properties->AddRef();
+
+ return PdxlnAppendTableScan(pexprRelational, colref_array,
+ pdrgpdsBaseTables, pexprScalar, dxl_properties);
+ }
+ case COperator::EopPhysicalParallelAppendTableScan:
+ {
+ dxl_properties->AddRef();
+
+ return PdxlnParallelAppendTableScan(pexprRelational, colref_array,
+ pdrgpdsBaseTables, pexprScalar, dxl_properties);
+ }
default:
{
return PdxlnResultFromFilter(pexprFilter, colref_array,
@@ -2944,6 +3309,7 @@ CTranslatorExprToDXL::PdxlnAppend(CExpression *pexprUnionAll,
CPhysicalUnionAll::PopConvert(pexprUnionAll->Pop());
CColRefArray *pdrgpcrOutputAll = popUnionAll->PdrgpcrOutput();
CColRefSet *reqdCols = pexprUnionAll->Prpp()->PcrsRequired();
+ m_isSet = true;
CDXLPhysicalAppend *dxl_op =
GPOS_NEW(m_mp) CDXLPhysicalAppend(m_mp, false, false);
@@ -3020,6 +3386,7 @@ CTranslatorExprToDXL::PdxlnAppend(CExpression *pexprUnionAll,
requiredInput->Release();
}
reqd_col_positions->Release();
+ m_isSet = false;
return pdxlnAppend;
}
@@ -4707,6 +5074,7 @@ CTranslatorExprToDXL::PdxlnResultFromNLJoinOuter(
// dynamic scan, thus we should not replace the filter.
case EdxlopPhysicalSequence:
case EdxlopPhysicalAppend:
+ case EdxlopPhysicalParallelAppend:
{
dxl_properties->AddRef();
GPOS_ASSERT(nullptr != pexprOuterChildRelational->Prpp());
diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2AppendTableScan.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2AppendTableScan.cpp
new file mode 100644
index 00000000000..8fc244e76ce
--- /dev/null
+++ b/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2AppendTableScan.cpp
@@ -0,0 +1,130 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CXformDynamicGet2AppendTableScan.cpp
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2AppendTableScan.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "gpopt/xforms/CXformDynamicGet2AppendTableScan.h"
+
+#include "gpos/base.h"
+
+#include "gpopt/hints/CHintUtils.h"
+#include "gpopt/metadata/CTableDescriptor.h"
+#include "gpopt/operators/CLogicalDynamicGet.h"
+#include "gpopt/operators/CPhysicalAppendTableScan.h"
+#include "gpopt/optimizer/COptimizerConfig.h"
+
+using namespace gpopt;
+
+
+//---------------------------------------------------------------------------
+// @function:
+// CXformDynamicGet2AppendTableScan::CXformDynamicGet2AppendTableScan
+//
+// @doc:
+// Ctor
+//
+//---------------------------------------------------------------------------
+CXformDynamicGet2AppendTableScan::CXformDynamicGet2AppendTableScan(
+ CMemoryPool *mp)
+ : CXformImplementation(
+ // pattern
+ GPOS_NEW(mp) CExpression(mp, GPOS_NEW(mp) CLogicalDynamicGet(mp)))
+{
+}
+
+// compute xform promise for a given expression handle
+CXform::EXformPromise
+CXformDynamicGet2AppendTableScan::Exfp(CExpressionHandle &exprhdl) const
+{
+ CLogicalDynamicGet *popGet = CLogicalDynamicGet::PopConvert(exprhdl.Pop());
+ // Do not run if contains foreign partitions, instead run CXformExpandDynamicGetWithForeignPartitions
+ if (popGet->ContainsForeignParts())
+ {
+ return CXform::ExfpNone;
+ }
+
+ if (!GPOS_FTRACE(EopttraceEnableParallelAppendScan))
+ return CXform::ExfpNone;
+
+ return CXform::ExfpHigh;
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CXformDynamicGet2AppendTableScan::Transform
+//
+// @doc:
+// Actual transformation
+//
+//---------------------------------------------------------------------------
+void
+CXformDynamicGet2AppendTableScan::Transform(CXformContext *pxfctxt,
+ CXformResult *pxfres,
+ CExpression *pexpr) const
+{
+ GPOS_ASSERT(nullptr != pxfctxt);
+ GPOS_ASSERT(FPromising(pxfctxt->Pmp(), this, pexpr));
+ GPOS_ASSERT(FCheckPattern(pexpr));
+
+ CLogicalDynamicGet *popGet = CLogicalDynamicGet::PopConvert(pexpr->Pop());
+
+ if (!CHintUtils::SatisfiesPlanHints(
+ popGet,
+ COptCtxt::PoctxtFromTLS()->GetOptimizerConfig()->GetPlanHint()))
+ {
+ return;
+ }
+
+ 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();
+
+ CColRef2dArray *pdrgpdrgpcrPart = popGet->PdrgpdrgpcrPart();
+ pdrgpdrgpcrPart->AddRef();
+
+ popGet->GetPartitionMdids()->AddRef();
+ popGet->GetRootColMappingPerPart()->AddRef();
+
+ // create alternative expression
+ CExpression *pexprAlt = GPOS_NEW(mp) CExpression(
+ mp, GPOS_NEW(mp) CPhysicalAppendTableScan(
+ mp, pname, ptabdesc, popGet->UlOpId(), popGet->ScanId(),
+ pdrgpcrOutput, pdrgpdrgpcrPart, popGet->GetPartitionMdids(),
+ popGet->GetRootColMappingPerPart()));
+ // add alternative to transformation result
+ pxfres->Add(pexprAlt);
+}
+
+
+// EOF
\ No newline at end of file
diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2ParallelAppendTableScan.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2ParallelAppendTableScan.cpp
new file mode 100644
index 00000000000..4b9cbacf87a
--- /dev/null
+++ b/src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2ParallelAppendTableScan.cpp
@@ -0,0 +1,184 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CXformDynamicGet2ParallelAppendTableScan.cpp
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/src/xforms/CXformDynamicGet2ParallelAppendTableScan.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "gpopt/xforms/CXformDynamicGet2ParallelAppendTableScan.h"
+
+#include "gpos/base.h"
+
+#include "gpopt/hints/CHintUtils.h"
+#include "gpopt/metadata/CTableDescriptor.h"
+#include "gpopt/operators/CLogicalDynamicGet.h"
+#include "gpopt/operators/CPhysicalParallelAppendTableScan.h"
+#include "gpopt/optimizer/COptimizerConfig.h"
+#include "gpopt/xforms/CXformGet2ParallelTableScan.h"
+#include "gpopt/xforms/CXformUtils.h"
+
+// Forward declarations for gpdbwrappers functions
+namespace gpdb {
+bool IsParallelModeOK(void);
+}
+
+using namespace gpopt;
+
+// Use gpdbwrappers for parallel checks
+extern int max_parallel_workers_per_gather;
+
+//---------------------------------------------------------------------------
+// @function:
+// CXformDynamicGet2ParallelAppendTableScan::CXformDynamicGet2ParallelAppendTableScan
+//
+// @doc:
+// Ctor
+//
+//---------------------------------------------------------------------------
+CXformDynamicGet2ParallelAppendTableScan::CXformDynamicGet2ParallelAppendTableScan(
+ CMemoryPool *mp)
+ : CXformImplementation(
+ // pattern
+ GPOS_NEW(mp) CExpression(mp, GPOS_NEW(mp) CLogicalDynamicGet(mp)))
+{
+}
+
+// compute xform promise for a given expression handle
+CXform::EXformPromise
+CXformDynamicGet2ParallelAppendTableScan::Exfp(CExpressionHandle &exprhdl) const
+{
+ // Check if parallel plans are enabled in context and parallel processing in safe
+ if (!gpdb::IsParallelModeOK())
+ {
+ return CXform::ExfpNone;
+ }
+
+ if (!GPOS_FTRACE(EopttraceEnableParallelAppendScan))
+ return CXform::ExfpNone;
+
+ // Same guard as CXformGet2ParallelTableScan: set operations, CTEs and
+ // Sequence are not worker-aware. If a parallel append scan ends up under
+ // one of them, a serial sibling in the same slice is executed by every
+ // worker and its rows are duplicated.
+ if (CXformGet2ParallelTableScan::FHasParallelIncompatibleOps(exprhdl))
+ {
+ return CXform::ExfpNone;
+ }
+
+ CLogicalDynamicGet *popGet = CLogicalDynamicGet::PopConvert(exprhdl.Pop());
+ CTableDescriptor *ptabdesc = popGet->Ptabdesc();
+
+ // Don't use parallel append for replicated tables
+ if (ptabdesc->GetRelDistribution() == IMDRelation::EreldistrReplicated ||
+ ptabdesc->GetRelDistribution() == IMDRelation::EreldistrMasterOnly ||
+ COptCtxt::PoctxtFromTLS()->HasReplicatedTables())
+ {
+ return CXform::ExfpNone;
+ }
+
+ // Do not run if contains foreign partitions, instead run CXformExpandDynamicGetWithForeignPartitions
+ if (popGet->ContainsForeignParts())
+ {
+ return CXform::ExfpNone;
+ }
+
+ return CXform::ExfpHigh;
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CXformDynamicGet2ParallelAppendTableScan::Transform
+//
+// @doc:
+// Actual transformation
+//
+//---------------------------------------------------------------------------
+void
+CXformDynamicGet2ParallelAppendTableScan::Transform(CXformContext *pxfctxt,
+ CXformResult *pxfres,
+ CExpression *pexpr) const
+{
+ GPOS_ASSERT(nullptr != pxfctxt);
+ GPOS_ASSERT(FPromising(pxfctxt->Pmp(), this, pexpr));
+ GPOS_ASSERT(FCheckPattern(pexpr));
+
+ CLogicalDynamicGet *popGet = CLogicalDynamicGet::PopConvert(pexpr->Pop());
+
+ if (!CHintUtils::SatisfiesPlanHints(
+ popGet,
+ COptCtxt::PoctxtFromTLS()->GetOptimizerConfig()->GetPlanHint()))
+ {
+ return;
+ }
+
+ 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();
+
+ CColRef2dArray *pdrgpdrgpcrPart = popGet->PdrgpdrgpcrPart();
+ pdrgpdrgpcrPart->AddRef();
+
+ popGet->GetPartitionMdids()->AddRef();
+ popGet->GetRootColMappingPerPart()->AddRef();
+
+ // Determine parallel workers degree
+ // Priority: table-level parallel_workers setting > GUC max_parallel_workers_per_gather > default
+ ULONG ulParallelWorkers = 2; // default
+
+ // 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) CPhysicalParallelAppendTableScan(
+ mp, pname, ptabdesc, popGet->UlOpId(), popGet->ScanId(),
+ pdrgpcrOutput, pdrgpdrgpcrPart, popGet->GetPartitionMdids(),
+ popGet->GetRootColMappingPerPart(), ulParallelWorkers));
+ // add alternative to transformation result
+ pxfres->Add(pexprAlt);
+}
+
+
+// EOF
diff --git a/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp b/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp
index 7ffe39774dc..5468846af81 100644
--- a/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp
+++ b/src/backend/gporca/libgpopt/src/xforms/CXformFactory.cpp
@@ -147,6 +147,8 @@ CXformFactory::Instantiate()
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) CXformDynamicGet2AppendTableScan(m_mp));
+ Add(GPOS_NEW(m_mp) CXformDynamicGet2ParallelAppendTableScan(m_mp));
Add(GPOS_NEW(m_mp) CXformDynamicIndexGet2DynamicIndexScan(m_mp));
Add(GPOS_NEW(m_mp) CXformImplementSequence(m_mp));
Add(GPOS_NEW(m_mp) CXformImplementConstTableGet(m_mp));
diff --git a/src/backend/gporca/libgpopt/src/xforms/Makefile b/src/backend/gporca/libgpopt/src/xforms/Makefile
index 66da51e28bb..d467347cd02 100644
--- a/src/backend/gporca/libgpopt/src/xforms/Makefile
+++ b/src/backend/gporca/libgpopt/src/xforms/Makefile
@@ -26,6 +26,8 @@ OBJS = CDecorrelator.o \
CXformDifference2LeftAntiSemiJoin.o \
CXformDifferenceAll2LeftAntiSemiJoin.o \
CXformDynamicGet2DynamicTableScan.o \
+ CXformDynamicGet2AppendTableScan.o \
+ CXformDynamicGet2ParallelAppendTableScan.o \
CXformDynamicIndexOnlyGet2DynamicIndexOnlyScan.o \
CXformDynamicIndexGet2DynamicIndexScan.o \
CXformEagerAgg.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 e583e9974a3..923ea162cfa 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperator.h
@@ -145,6 +145,7 @@ enum Edxlopid
EdxlopPhysicalAgg,
EdxlopPhysicalSort,
EdxlopPhysicalAppend,
+ EdxlopPhysicalParallelAppend,
EdxlopPhysicalMaterialize,
EdxlopPhysicalDynamicForeignScan,
EdxlopPhysicalSequence,
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 568240a5c98..5ffb517fb55 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLOperatorFactory.h
@@ -177,6 +177,10 @@ public:
static CDXLPhysical *MakeDXLAppend(CDXLMemoryManager *dxl_memory_manager,
const Attributes &attrs);
+ // create a parallel append operator
+ static CDXLPhysical *MakeDXLParallelAppend(CDXLMemoryManager *dxl_memory_manager,
+ const Attributes &attrs);
+
// create a limit operator
static CDXLPhysical *MakeDXLLimit(CDXLMemoryManager *dxl_memory_manager,
const Attributes &attrs);
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalAppend.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalAppend.h
index 2b547015447..c53d1466ce2 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalAppend.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalAppend.h
@@ -64,7 +64,7 @@ public:
// ctor/dtor
CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget, BOOL fIsZapped);
- // ctor for partitioned table scan
+ // ctor for paritioned table scan
CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget, BOOL fIsZapped,
ULONG scan_id, CDXLTableDescr *dxl_table_desc,
ULongPtrArray *selector_ids);
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelAppend.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelAppend.h
new file mode 100644
index 00000000000..54bd7c80cc4
--- /dev/null
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelAppend.h
@@ -0,0 +1,143 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CDXLPhysicalParallelAppend.h
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libnaucrates/include/naucrates/dxl/operators/CDXLPhysicalParallelAppend.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef GPDXL_CDXLPhysicalParallelAppend_H
+#define GPDXL_CDXLPhysicalParallelAppend_H
+
+#include "gpos/base.h"
+
+#include "naucrates/dxl/operators/CDXLPhysicalAppend.h"
+
+namespace gpdxl
+{
+using namespace gpos;
+
+//---------------------------------------------------------------------------
+// @class:
+// CDXLPhysicalParallelAppend
+//
+// @doc:
+// Class for representing DXL parallel append operators
+//
+//---------------------------------------------------------------------------
+class CDXLPhysicalParallelAppend : public CDXLPhysical
+{
+private:
+ // is the append node used in an update/delete statement
+ BOOL m_used_in_upd_del = false;
+
+ // TODO: - Apr 12, 2011; find a better name (and comments) for this variable
+ BOOL m_is_zapped = false;
+
+ // scan id from the CPhysicalDynamicTableScan (a.k.a part_index_id)
+ // when m_scan_id != gpos::ulong_max
+ ULONG m_scan_id = gpos::ulong_max;
+
+ // table descr of the root partitioned table (when translated from a CPhysicalDynamicTableScan)
+ CDXLTableDescr *m_dxl_table_descr = nullptr;
+
+ ULongPtrArray *m_selector_ids = nullptr;
+
+ // number of parallel workers
+ ULONG m_ulParallelWorkers;
+
+public:
+ CDXLPhysicalParallelAppend(const CDXLPhysicalParallelAppend &) = delete;
+
+ //ctor
+ CDXLPhysicalParallelAppend(CMemoryPool *mp, BOOL fIsTarget, BOOL fIsZapped, ULONG ulParallelWorkers);
+
+ // ctor for paritioned table scan
+ CDXLPhysicalParallelAppend(CMemoryPool *mp, BOOL fIsTarget, BOOL fIsZapped,
+ ULONG scan_id, CDXLTableDescr *dxl_table_desc,
+ ULongPtrArray *selector_ids, ULONG ulParallelWorkers);
+
+ // dtor
+ ~CDXLPhysicalParallelAppend() override;
+
+ // accessors
+ Edxlopid GetDXLOperator() const override;
+ const CWStringConst *GetOpNameStr() const override;
+
+ BOOL IsUsedInUpdDel() const;
+ BOOL IsZapped() const;
+
+ CDXLTableDescr *
+ GetDXLTableDesc() const
+ {
+ return m_dxl_table_descr;
+ }
+
+ void
+ SetDXLTableDesc(CDXLTableDescr *dxl_table_desc)
+ {
+ m_dxl_table_descr = dxl_table_desc;
+ }
+
+ ULONG
+ GetScanId() const
+ {
+ return m_scan_id;
+ }
+
+ const ULongPtrArray *
+ GetSelectorIds() const
+ {
+ return m_selector_ids;
+ }
+
+ // 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 CDXLPhysicalParallelAppend *
+ Cast(CDXLOperator *dxl_op)
+ {
+ GPOS_ASSERT(nullptr != dxl_op);
+ GPOS_ASSERT(EdxlopPhysicalParallelAppend == dxl_op->GetDXLOperator());
+
+ return dynamic_cast<CDXLPhysicalParallelAppend *>(dxl_op);
+ }
+
+#ifdef GPOS_DEBUG
+ // checks whether the operator has valid structure, i.e. number and
+ // types of child nodes
+ void AssertValid(const CDXLNode *, BOOL validate_children) const override;
+#endif // GPOS_DEBUG
+}; // class CDXLPhysicalParallelAppend
+
+} // namespace gpdxl
+#endif // !GPDXL_CDXLPhysicalParallelAppend_H
+
+// EOF
\ No newline at end of file
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerFactory.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerFactory.h
index 17abeca8fe1..c610aee6056 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerFactory.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerFactory.h
@@ -336,6 +336,11 @@ private:
CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
CParseHandlerBase *parse_handler_root);
+ // construct a parallel append parse handler
+ static CParseHandlerBase *CreateParallelAppendParseHandler(
+ CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
+ CParseHandlerBase *parse_handler_root);
+
// construct a materialize parse handler
static CParseHandlerBase *CreateMaterializeParseHandler(
CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerParallelAppend.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerParallelAppend.h
new file mode 100644
index 00000000000..4fe5157c5ce
--- /dev/null
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerParallelAppend.h
@@ -0,0 +1,84 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CParseHandlerParallelAppend.h
+ * SAX parse handler class for parsing ParallelAppend operator nodes.
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/CParseHandlerParallelAppend.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef GPDXL_CParseHandlerParallelAppend_H
+#define GPDXL_CParseHandlerParallelAppend_H
+
+#include "gpos/base.h"
+
+#include "naucrates/dxl/operators/CDXLPhysicalParallelAppend.h"
+#include "naucrates/dxl/parser/CParseHandlerPhysicalOp.h"
+
+namespace gpdxl
+{
+using namespace gpos;
+
+XERCES_CPP_NAMESPACE_USE
+
+//---------------------------------------------------------------------------
+// @class:
+// CParseHandlerParallelAppend
+//
+// @doc:
+// Parse handler for ParallelAppend operator
+//
+//---------------------------------------------------------------------------
+class CParseHandlerParallelAppend : public CParseHandlerPhysicalOp
+{
+private:
+ CDXLPhysicalParallelAppend *m_dxl_op;
+
+ // set up initial handlers
+ void SetupInitialHandlers(const Attributes &attrs);
+
+ // process the start of an element
+ void StartElement(
+ const XMLCh *const element_uri, // URI of element's namespace
+ const XMLCh *const element_local_name, // local part of element's name
+ const XMLCh *const element_qname, // element's qname
+ const Attributes &attrs // element's attributes
+ ) override;
+
+ // process the end of an element
+ void EndElement(
+ const XMLCh *const element_uri, // URI of element's namespace
+ const XMLCh *const element_local_name, // local part of element's name
+ const XMLCh *const element_qname // element's qname
+ ) override;
+
+public:
+ CParseHandlerParallelAppend(const CParseHandlerParallelAppend &) = delete;
+
+ // ctor/dtor
+ CParseHandlerParallelAppend(CMemoryPool *mp,
+ CParseHandlerManager *parse_handler_mgr,
+ CParseHandlerBase *parse_handler_root);
+};
+} // namespace gpdxl
+#endif // GPDXL_CParseHandlerParallelAppend_H
+
+// EOF
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/parsehandlers.h b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/parsehandlers.h
index d994fc55a61..f6438554946 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/parsehandlers.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/parser/parsehandlers.h
@@ -14,6 +14,7 @@
#include "naucrates/dxl/parser/CParseHandlerAgg.h"
#include "naucrates/dxl/parser/CParseHandlerAppend.h"
+#include "naucrates/dxl/parser/CParseHandlerParallelAppend.h"
#include "naucrates/dxl/parser/CParseHandlerArray.h"
#include "naucrates/dxl/parser/CParseHandlerAssert.h"
#include "naucrates/dxl/parser/CParseHandlerBroadcastMotion.h"
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 9c5d89fe8ae..9d4fd21111e 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/dxl/xml/dxltokens.h
@@ -116,6 +116,7 @@ enum Edxltoken
EdxltokenPhysicalResult,
EdxltokenPhysicalAggregate,
EdxltokenPhysicalAppend,
+ EdxltokenPhysicalParallelAppend,
EdxltokenPhysicalMaterialize,
EdxltokenPhysicalDynamicForeignScan,
EdxltokenPhysicalSequence,
diff --git a/src/backend/gporca/libnaucrates/include/naucrates/traceflags/traceflags.h b/src/backend/gporca/libnaucrates/include/naucrates/traceflags/traceflags.h
index 8a18ace986a..93b30c38ea2 100644
--- a/src/backend/gporca/libnaucrates/include/naucrates/traceflags/traceflags.h
+++ b/src/backend/gporca/libnaucrates/include/naucrates/traceflags/traceflags.h
@@ -188,6 +188,9 @@ enum EOptTraceFlag
// Eager Agg
EopttraceEnableEagerAgg = 103030,
+ // Enable parallel append scan
+ EopttraceEnableParallelAppendScan = 103031,
+
// ExpandFullJoin
EopttraceExpandFullJoin = 103032,
diff --git a/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp b/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp
index a4d4914b731..94cd44bc49e 100644
--- a/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp
+++ b/src/backend/gporca/libnaucrates/src/operators/CDXLOperatorFactory.cpp
@@ -31,6 +31,7 @@
#include "naucrates/dxl/operators/CDXLPhysicalParallelTableScan.h"
#include "naucrates/dxl/operators/CDXLPhysicalAgg.h"
#include "naucrates/dxl/operators/CDXLPhysicalAppend.h"
+#include "naucrates/dxl/operators/CDXLPhysicalParallelAppend.h"
#include "naucrates/dxl/operators/CDXLPhysicalBroadcastMotion.h"
#include "naucrates/dxl/operators/CDXLPhysicalGatherMotion.h"
#include "naucrates/dxl/operators/CDXLPhysicalHashJoin.h"
@@ -408,6 +409,48 @@ CDXLOperatorFactory::MakeDXLAppend(CDXLMemoryManager *dxl_memory_manager,
nullptr, selector_ids);
}
+//---------------------------------------------------------------------------
+// @function:
+// CDXLOperatorFactory::MakeDXLParallelAppend
+// @doc:
+// Construct a Parallel Append operator
+//
+//---------------------------------------------------------------------------
+CDXLPhysical *
+CDXLOperatorFactory::MakeDXLParallelAppend(CDXLMemoryManager *dxl_memory_manager,
+ const Attributes &attrs)
+{
+ // get the memory pool from the memory manager
+ CMemoryPool *mp = dxl_memory_manager->Pmp();
+
+ BOOL is_target = ExtractConvertAttrValueToBool(dxl_memory_manager, attrs,
+ EdxltokenAppendIsTarget,
+ EdxltokenPhysicalParallelAppend);
+
+ BOOL is_zapped = ExtractConvertAttrValueToBool(dxl_memory_manager, attrs,
+ EdxltokenAppendIsZapped,
+ EdxltokenPhysicalParallelAppend);
+
+ ULONG scan_id = ExtractConvertAttrValueToUlong(dxl_memory_manager, attrs, EdxltokenPartIndexId,
+ EdxltokenPhysicalParallelAppend, true /* is_optional */,
+ gpos::ulong_max /* default_value */);
+
+ ULONG parallel_workers = ExtractConvertAttrValueToUlong(dxl_memory_manager, attrs, EdxltokenParallelWorkers,
+ EdxltokenPhysicalParallelAppend, true,
+ 0);
+
+ ULongPtrArray *selector_ids = nullptr;
+ if (scan_id != gpos::ulong_max)
+ {
+ selector_ids = ExtractConvertValuesToArray(dxl_memory_manager, attrs,
+ EdxltokenSelectorIds,
+ EdxltokenPhysicalParallelAppend);
+ }
+
+ return GPOS_NEW(mp) CDXLPhysicalParallelAppend(mp, is_target, is_zapped, scan_id,
+ nullptr, selector_ids, parallel_workers);
+}
+
//---------------------------------------------------------------------------
// @function:
// CDXLOperatorFactory::MakeDXLLimit
diff --git a/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp
index 8e90a279cd8..adbdbd9bec1 100644
--- a/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp
+++ b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp
@@ -34,8 +34,10 @@ CDXLPhysicalAppend::CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget,
{
}
-CDXLPhysicalAppend::CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget,
- BOOL fIsZapped, ULONG scan_id,
+CDXLPhysicalAppend::CDXLPhysicalAppend(CMemoryPool *mp,
+ BOOL fIsTarget,
+ BOOL fIsZapped,
+ ULONG scan_id,
CDXLTableDescr *dxl_table_desc,
ULongPtrArray *selector_ids)
: CDXLPhysical(mp),
diff --git a/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelAppend.cpp
similarity index 57%
copy from src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp
copy to src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelAppend.cpp
index 8e90a279cd8..55289774440 100644
--- a/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalAppend.cpp
+++ b/src/backend/gporca/libnaucrates/src/operators/CDXLPhysicalParallelAppend.cpp
@@ -1,15 +1,31 @@
-//---------------------------------------------------------------------------
-// Greenplum Database
-// Copyright (C) 2011 Greenplum, Inc.
-//
-// @filename:
-// CDXLPhysicalAppend.cpp
-//
-// @doc:
-// Implementation of DXL physical Append operator
-//---------------------------------------------------------------------------
-
-#include "naucrates/dxl/operators/CDXLPhysicalAppend.h"
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CPhysicalParallelAppend.cpp
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libgpopt/src/operators/CPhysicalParallelAppend.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "naucrates/dxl/operators/CDXLPhysicalParallelAppend.h"
#include "gpos/common/CBitSetIter.h"
@@ -22,32 +38,36 @@ using namespace gpdxl;
//---------------------------------------------------------------------------
// @function:
-// CDXLPhysicalAppend::CDXLPhysicalAppend
+// CDXLPhysicalParallelAppend::CDXLPhysicalParallelAppend
//
// @doc:
// Constructor
//
//---------------------------------------------------------------------------
-CDXLPhysicalAppend::CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget,
- BOOL fIsZapped)
- : CDXLPhysical(mp), m_used_in_upd_del(fIsTarget), m_is_zapped(fIsZapped)
+CDXLPhysicalParallelAppend::CDXLPhysicalParallelAppend(CMemoryPool *mp, BOOL fIsTarget,
+ BOOL fIsZapped, ULONG ulParallelWorkers)
+ : CDXLPhysical(mp), m_used_in_upd_del(fIsTarget), m_is_zapped(fIsZapped), m_ulParallelWorkers(ulParallelWorkers)
{
}
-CDXLPhysicalAppend::CDXLPhysicalAppend(CMemoryPool *mp, BOOL fIsTarget,
- BOOL fIsZapped, ULONG scan_id,
- CDXLTableDescr *dxl_table_desc,
- ULongPtrArray *selector_ids)
+CDXLPhysicalParallelAppend::CDXLPhysicalParallelAppend(CMemoryPool *mp,
+ BOOL fIsTarget,
+ BOOL fIsZapped,
+ ULONG scan_id,
+ CDXLTableDescr *dxl_table_desc,
+ ULongPtrArray *selector_ids,
+ ULONG ulParallelWorkers)
: CDXLPhysical(mp),
m_used_in_upd_del(fIsTarget),
m_is_zapped(fIsZapped),
m_scan_id(scan_id),
m_dxl_table_descr(dxl_table_desc),
- m_selector_ids(selector_ids)
+ m_selector_ids(selector_ids),
+ m_ulParallelWorkers(ulParallelWorkers)
{
}
-CDXLPhysicalAppend::~CDXLPhysicalAppend()
+CDXLPhysicalParallelAppend::~CDXLPhysicalParallelAppend()
{
CRefCount::SafeRelease(m_dxl_table_descr);
CRefCount::SafeRelease(m_selector_ids);
@@ -55,72 +75,42 @@ CDXLPhysicalAppend::~CDXLPhysicalAppend()
//---------------------------------------------------------------------------
// @function:
-// CDXLPhysicalAppend::GetDXLOperator
+// CDXLPhysicalParallelAppend::GetDXLOperator
//
// @doc:
// Operator type
//
//---------------------------------------------------------------------------
Edxlopid
-CDXLPhysicalAppend::GetDXLOperator() const
+CDXLPhysicalParallelAppend::GetDXLOperator() const
{
- return EdxlopPhysicalAppend;
+ return EdxlopPhysicalParallelAppend;
}
-
//---------------------------------------------------------------------------
// @function:
-// CDXLPhysicalAppend::GetOpNameStr
+// CDXLPhysicalParallelAppend::GetOpNameStr
//
// @doc:
// Operator name
//
//---------------------------------------------------------------------------
const CWStringConst *
-CDXLPhysicalAppend::GetOpNameStr() const
-{
- return CDXLTokens::GetDXLTokenStr(EdxltokenPhysicalAppend);
-}
-
-//---------------------------------------------------------------------------
-// @function:
-// CDXLPhysicalAppend::IsUsedInUpdDel
-//
-// @doc:
-// Is the append node updating a target relation
-//
-//---------------------------------------------------------------------------
-BOOL
-CDXLPhysicalAppend::IsUsedInUpdDel() const
-{
- return m_used_in_upd_del;
-}
-
-//---------------------------------------------------------------------------
-// @function:
-// CDXLPhysicalAppend::IsZapped
-//
-// @doc:
-// Is the append node zapped
-//
-//---------------------------------------------------------------------------
-BOOL
-CDXLPhysicalAppend::IsZapped() const
+CDXLPhysicalParallelAppend::GetOpNameStr() const
{
- return m_is_zapped;
+ return CDXLTokens::GetDXLTokenStr(EdxltokenPhysicalParallelAppend);
}
//---------------------------------------------------------------------------
// @function:
-// CDXLPhysicalAppend::SerializeToDXL
+// CDXLPhysicalParallelAppend::SerializeToDXL
//
// @doc:
// Serialize operator in DXL format
//
//---------------------------------------------------------------------------
void
-CDXLPhysicalAppend::SerializeToDXL(CXMLSerializer *xml_serializer,
- const CDXLNode *dxlnode) const
+CDXLPhysicalParallelAppend::SerializeToDXL(CXMLSerializer *xml_serializer, const CDXLNode *dxlnode) const
{
const CWStringConst *element_name = GetOpNameStr();
@@ -131,6 +121,8 @@ CDXLPhysicalAppend::SerializeToDXL(CXMLSerializer *xml_serializer,
CDXLTokens::GetDXLTokenStr(EdxltokenAppendIsTarget), m_used_in_upd_del);
xml_serializer->AddAttribute(
CDXLTokens::GetDXLTokenStr(EdxltokenAppendIsZapped), m_is_zapped);
+ xml_serializer->AddAttribute(
+ CDXLTokens::GetDXLTokenStr(EdxltokenParallelWorkers), m_ulParallelWorkers);
if (m_scan_id != gpos::ulong_max)
{
@@ -153,7 +145,6 @@ CDXLPhysicalAppend::SerializeToDXL(CXMLSerializer *xml_serializer,
m_dxl_table_descr->SerializeToDXL(xml_serializer);
}
-
// serialize children
dxlnode->SerializeChildrenToDXL(xml_serializer);
@@ -161,7 +152,6 @@ CDXLPhysicalAppend::SerializeToDXL(CXMLSerializer *xml_serializer,
CDXLTokens::GetDXLTokenStr(EdxltokenNamespacePrefix), element_name);
}
-
#ifdef GPOS_DEBUG
//---------------------------------------------------------------------------
// @function:
@@ -172,7 +162,7 @@ CDXLPhysicalAppend::SerializeToDXL(CXMLSerializer *xml_serializer,
//
//---------------------------------------------------------------------------
void
-CDXLPhysicalAppend::AssertValid(const CDXLNode *dxlnode,
+CDXLPhysicalParallelAppend::AssertValid(const CDXLNode *dxlnode,
BOOL validate_children) const
{
// assert proj list and filter are valid
@@ -194,4 +184,4 @@ CDXLPhysicalAppend::AssertValid(const CDXLNode *dxlnode,
}
#endif // GPOS_DEBUG
-// EOF
+// EOF
diff --git a/src/backend/gporca/libnaucrates/src/operators/Makefile b/src/backend/gporca/libnaucrates/src/operators/Makefile
index 1d78142feab..599d3e8ca71 100644
--- a/src/backend/gporca/libnaucrates/src/operators/Makefile
+++ b/src/backend/gporca/libnaucrates/src/operators/Makefile
@@ -51,6 +51,7 @@ OBJS = CDXLColDescr.o \
CDXLPhysicalAbstractBitmapScan.o \
CDXLPhysicalAgg.o \
CDXLPhysicalAppend.o \
+ CDXLPhysicalParallelAppend.o \
CDXLPhysicalAssert.o \
CDXLPhysicalBitmapTableScan.o \
CDXLPhysicalBroadcastMotion.o \
diff --git a/src/backend/gporca/libnaucrates/src/parser/CParseHandlerFactory.cpp b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerFactory.cpp
index 3dd1ce8a146..a5715b61e98 100644
--- a/src/backend/gporca/libnaucrates/src/parser/CParseHandlerFactory.cpp
+++ b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerFactory.cpp
@@ -111,6 +111,7 @@ CParseHandlerFactory::Init(CMemoryPool *mp)
{EdxltokenPhysicalLimit, &CreateLimitParseHandler},
{EdxltokenPhysicalSort, &CreateSortParseHandler},
{EdxltokenPhysicalAppend, &CreateAppendParseHandler},
+ {EdxltokenPhysicalParallelAppend, &CreateParallelAppendParseHandler},
{EdxltokenPhysicalMaterialize, &CreateMaterializeParseHandler},
{EdxltokenPhysicalDynamicTableScan, &CreateDTSParseHandler},
{EdxltokenPhysicalDynamicIndexScan, &CreateDynamicIdxScanParseHandler},
@@ -821,6 +822,16 @@ CParseHandlerFactory::CreateAppendParseHandler(
CParseHandlerAppend(mp, parse_handler_mgr, parse_handler_root);
}
+// creates a parse handler for parsing a parallel append operator
+CParseHandlerBase *
+CParseHandlerFactory::CreateParallelAppendParseHandler(
+ CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
+ CParseHandlerBase *parse_handler_root)
+{
+ return GPOS_NEW(mp)
+ CParseHandlerParallelAppend(mp, parse_handler_mgr, parse_handler_root);
+}
+
// creates a parse handler for parsing a materialize operator
CParseHandlerBase *
CParseHandlerFactory::CreateMaterializeParseHandler(
diff --git a/src/backend/gporca/libnaucrates/src/parser/CParseHandlerParallelAppend.cpp b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerParallelAppend.cpp
new file mode 100644
index 00000000000..c933feb2ee6
--- /dev/null
+++ b/src/backend/gporca/libnaucrates/src/parser/CParseHandlerParallelAppend.cpp
@@ -0,0 +1,237 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * CParseHandlerParallelAppend.cpp
+ * Implementation of the SAX parse handler class for parsing ParallelAppend operator.
+ *
+ * IDENTIFICATION
+ * src/backend/gporca/libnaucrates/src/parser/CParseHandlerParallelAppend.cpp
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "naucrates/dxl/parser/CParseHandlerParallelAppend.h"
+
+#include "naucrates/dxl/operators/CDXLOperatorFactory.h"
+#include "naucrates/dxl/parser/CParseHandlerFactory.h"
+#include "naucrates/dxl/parser/CParseHandlerFilter.h"
+#include "naucrates/dxl/parser/CParseHandlerProjList.h"
+#include "naucrates/dxl/parser/CParseHandlerProperties.h"
+#include "naucrates/dxl/parser/CParseHandlerScalarOp.h"
+#include "naucrates/dxl/parser/CParseHandlerTableDescr.h"
+#include "naucrates/dxl/parser/CParseHandlerUtils.h"
+
+using namespace gpdxl;
+
+
+XERCES_CPP_NAMESPACE_USE
+
+//---------------------------------------------------------------------------
+// @function:
+// CParseHandlerParallelAppend::CParseHandlerParallelAppend
+//
+// @doc:
+// Constructor
+//
+//---------------------------------------------------------------------------
+CParseHandlerParallelAppend::CParseHandlerParallelAppend(
+ CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
+ CParseHandlerBase *parse_handler_root)
+ : CParseHandlerPhysicalOp(mp, parse_handler_mgr, parse_handler_root),
+ m_dxl_op(nullptr)
+{
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CParseHandlerParallelAppend::SetupInitialHandlers
+//
+// @doc:
+// Setup initial set of handlers for parallel append node
+//
+//---------------------------------------------------------------------------
+void
+CParseHandlerParallelAppend::SetupInitialHandlers(const Attributes &attrs)
+{
+ // seeing a result tag
+ GPOS_ASSERT(m_dxl_op == nullptr &&
+ "ParallelAppend dxl node should not have been created yet");
+ GPOS_ASSERT(this->Length() == 0 &&
+ "No handlers should have been added yet");
+
+ m_dxl_op = (CDXLPhysicalParallelAppend *) CDXLOperatorFactory::MakeDXLParallelAppend(
+ m_parse_handler_mgr->GetDXLMemoryManager(), attrs);
+
+ // parse handler for the filter
+ CParseHandlerBase *filter_parse_handler =
+ CParseHandlerFactory::GetParseHandler(
+ m_mp, CDXLTokens::XmlstrToken(EdxltokenScalarFilter),
+ m_parse_handler_mgr, this);
+ m_parse_handler_mgr->ActivateParseHandler(filter_parse_handler);
+
+ // parse handler for the table desc
+ CParseHandlerBase *table_descr_parse_handler = nullptr;
+ if (m_dxl_op->GetScanId() != gpos::ulong_max)
+ {
+ // parse handler for table descriptor
+ table_descr_parse_handler = CParseHandlerFactory::GetParseHandler(
+ m_mp, CDXLTokens::XmlstrToken(EdxltokenTableDescr),
+ m_parse_handler_mgr, this);
+ m_parse_handler_mgr->ActivateParseHandler(table_descr_parse_handler);
+ }
+
+ // parse handler for the proj list
+ CParseHandlerBase *proj_list_parse_handler =
+ CParseHandlerFactory::GetParseHandler(
+ m_mp, CDXLTokens::XmlstrToken(EdxltokenScalarProjList),
+ m_parse_handler_mgr, this);
+ m_parse_handler_mgr->ActivateParseHandler(proj_list_parse_handler);
+
+ //parse handler for the properties of the operator
+ CParseHandlerBase *prop_parse_handler =
+ CParseHandlerFactory::GetParseHandler(
+ m_mp, CDXLTokens::XmlstrToken(EdxltokenProperties),
+ m_parse_handler_mgr, this);
+ m_parse_handler_mgr->ActivateParseHandler(prop_parse_handler);
+
+ this->Append(prop_parse_handler);
+ if (m_dxl_op->GetScanId() != gpos::ulong_max)
+ {
+ GPOS_ASSERT(nullptr != table_descr_parse_handler);
+ this->Append(table_descr_parse_handler);
+ }
+ this->Append(proj_list_parse_handler);
+ this->Append(filter_parse_handler);
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CParseHandlerParallelAppend::StartElement
+//
+// @doc:
+// Invoked by Xerces to process an opening tag
+//
+//---------------------------------------------------------------------------
+void
+CParseHandlerParallelAppend::StartElement(const XMLCh *const element_uri,
+ const XMLCh *const element_local_name,
+ const XMLCh *const element_qname,
+ const Attributes &attrs)
+{
+ if (0 == XMLString::compareString(
+ CDXLTokens::XmlstrToken(EdxltokenPhysicalParallelAppend),
+ element_local_name) &&
+ nullptr == m_dxl_op)
+ {
+ // open a root Append element
+ SetupInitialHandlers(attrs);
+ }
+ else if (nullptr != m_dxl_op)
+ {
+ // install a parse handler for a child node
+ CParseHandlerBase *child_parse_handler =
+ CParseHandlerFactory::GetParseHandler(
+ m_mp, CDXLTokens::XmlstrToken(EdxltokenPhysical),
+ m_parse_handler_mgr, this);
+ m_parse_handler_mgr->ActivateParseHandler(child_parse_handler);
+
+ this->Append(child_parse_handler);
+
+ child_parse_handler->startElement(element_uri, element_local_name,
+ element_qname, attrs);
+ }
+ else
+ {
+ CWStringDynamic *str = CDXLUtils::CreateDynamicStringFromXMLChArray(
+ m_parse_handler_mgr->GetDXLMemoryManager(), element_local_name);
+ GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiDXLUnexpectedTag,
+ str->GetBuffer());
+ }
+}
+
+//---------------------------------------------------------------------------
+// @function:
+// CParseHandlerParallelAppend::EndElement
+//
+// @doc:
+// Invoked by Xerces to process a closing tag
+//
+//---------------------------------------------------------------------------
+void
+CParseHandlerParallelAppend::EndElement(const XMLCh *const, // element_uri,
+ const XMLCh *const element_local_name,
+ const XMLCh *const // element_qname
+)
+{
+ if (0 != XMLString::compareString(
+ CDXLTokens::XmlstrToken(EdxltokenPhysicalParallelAppend),
+ element_local_name))
+ {
+ CWStringDynamic *str = CDXLUtils::CreateDynamicStringFromXMLChArray(
+ m_parse_handler_mgr->GetDXLMemoryManager(), element_local_name);
+ GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiDXLUnexpectedTag,
+ str->GetBuffer());
+ }
+
+ ULONG child_index = 0;
+ // construct node from the created child nodes
+ CParseHandlerProperties *prop_parse_handler =
+ dynamic_cast<CParseHandlerProperties *>((*this)[child_index++]);
+
+ if (m_dxl_op->GetScanId() != gpos::ulong_max)
+ {
+ CParseHandlerTableDescr *table_descr_parse_handler =
+ dynamic_cast<CParseHandlerTableDescr *>((*this)[child_index++]);
+ CDXLTableDescr *table_descr =
+ table_descr_parse_handler->GetDXLTableDescr();
+ table_descr->AddRef();
+ m_dxl_op->SetDXLTableDesc(table_descr);
+ }
+
+ CParseHandlerProjList *proj_list_parse_handler =
+ dynamic_cast<CParseHandlerProjList *>((*this)[child_index++]);
+ CParseHandlerFilter *filter_parse_handler =
+ dynamic_cast<CParseHandlerFilter *>((*this)[child_index++]);
+
+ m_dxl_node = GPOS_NEW(m_mp) CDXLNode(m_mp, m_dxl_op);
+ CParseHandlerUtils::SetProperties(m_dxl_node, prop_parse_handler);
+
+ // add constructed children
+ AddChildFromParseHandler(proj_list_parse_handler);
+ AddChildFromParseHandler(filter_parse_handler);
+
+ GPOS_ASSERT(3 <= this->Length());
+
+ const ULONG length = this->Length();
+ // an append node can have variable number of children: add them one by one from the respective parse handlers
+ for (; child_index < length; ++child_index)
+ {
+ CParseHandlerPhysicalOp *child_parse_handler =
+ dynamic_cast<CParseHandlerPhysicalOp *>((*this)[child_index]);
+ AddChildFromParseHandler(child_parse_handler);
+ }
+
+#ifdef GPOS_DEBUG
+ m_dxl_op->AssertValid(m_dxl_node, false /* validate_children */);
+#endif // GPOS_DEBUG
+
+ // deactivate handler
+ m_parse_handler_mgr->DeactivateHandler();
+}
+
+// EOF
diff --git a/src/backend/gporca/libnaucrates/src/parser/Makefile b/src/backend/gporca/libnaucrates/src/parser/Makefile
index d141742a0ff..fdf4fd5d30e 100644
--- a/src/backend/gporca/libnaucrates/src/parser/Makefile
+++ b/src/backend/gporca/libnaucrates/src/parser/Makefile
@@ -12,6 +12,7 @@ include $(top_srcdir)/src/backend/gporca/gporca.mk
OBJS = CParseHandlerAgg.o \
CParseHandlerAppend.o \
+ CParseHandlerParallelAppend.o \
CParseHandlerArray.o \
CParseHandlerAssert.o \
CParseHandlerBase.o \
diff --git a/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp b/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp
index 79fcc9f7d6a..c5804e9ccf5 100644
--- a/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp
+++ b/src/backend/gporca/libnaucrates/src/xml/dxltokens.cpp
@@ -137,6 +137,7 @@ CDXLTokens::Init(CMemoryPool *mp)
{EdxltokenPhysicalResult, GPOS_WSZ_LIT("Result")},
{EdxltokenPhysicalValuesScan, GPOS_WSZ_LIT("Values")},
{EdxltokenPhysicalAppend, GPOS_WSZ_LIT("Append")},
+ {EdxltokenPhysicalParallelAppend, GPOS_WSZ_LIT("ParallelAppend")},
{EdxltokenPhysicalMaterialize, GPOS_WSZ_LIT("Materialize")},
{EdxltokenPhysicalDynamicForeignScan,
GPOS_WSZ_LIT("DynamicForeignScan")},
diff --git a/src/backend/utils/misc/guc_gp.c b/src/backend/utils/misc/guc_gp.c
index ea632817910..18c5661599f 100644
--- a/src/backend/utils/misc/guc_gp.c
+++ b/src/backend/utils/misc/guc_gp.c
@@ -333,6 +333,7 @@ bool optimizer_enable_motion_gather;
bool optimizer_enable_motion_redistribute;
bool optimizer_enable_sort;
bool optimizer_enable_materialize;
+bool optimizer_enable_parallel_append;
bool optimizer_enable_partition_propagation;
bool optimizer_enable_partition_selection;
bool optimizer_enable_outerjoin_rewrite;
@@ -2272,6 +2273,16 @@ struct config_bool ConfigureNamesBool_gp[] =
true,
NULL, NULL, NULL
},
+ {
+ {"optimizer_enable_parallel_append", PGC_USERSET, DEVELOPER_OPTIONS,
+ gettext_noop("Enable parallel append for seq/bitmap/index scan in partition table."),
+ NULL,
+ GUC_NOT_IN_SAMPLE
+ },
+ &optimizer_enable_parallel_append,
+ false,
+ NULL, NULL, NULL
+ },
{
{"optimizer_enable_partition_propagation", PGC_USERSET, DEVELOPER_OPTIONS,
gettext_noop("Enable plans with Partition Propagation operators in the optimizer."),
diff --git a/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h b/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h
index 3cd5fc9d638..78f8aa52bd8 100644
--- a/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h
+++ b/src/include/gpopt/translate/CTranslatorDXLToPlStmt.h
@@ -361,6 +361,12 @@ private:
ctxt_translation_prev_siblings // translation contexts of previous siblings
);
+ Plan *TranslateDXLParallelAppend(
+ const CDXLNode *append_dxlnode, CDXLTranslateContext *output_context,
+ CDXLTranslationContextArray *
+ ctxt_translation_prev_siblings // translation contexts of previous siblings
+ );
+
Plan *TranslateDXLMaterialize(
const CDXLNode *materialize_dxlnode,
CDXLTranslateContext *output_context,
diff --git a/src/include/utils/guc.h b/src/include/utils/guc.h
index 4f6401fb43f..4f9767ea4f4 100644
--- a/src/include/utils/guc.h
+++ b/src/include/utils/guc.h
@@ -521,6 +521,7 @@ extern bool optimizer_enable_motion_gather;
extern bool optimizer_enable_motion_redistribute;
extern bool optimizer_enable_sort;
extern bool optimizer_enable_materialize;
+extern bool optimizer_enable_parallel_append;
extern bool optimizer_enable_partition_propagation;
extern bool optimizer_enable_partition_selection;
extern bool optimizer_enable_outerjoin_rewrite;
diff --git a/src/include/utils/unsync_guc_name.h b/src/include/utils/unsync_guc_name.h
index 02b04fdc5c4..f313e7dacc4 100644
--- a/src/include/utils/unsync_guc_name.h
+++ b/src/include/utils/unsync_guc_name.h
@@ -437,6 +437,7 @@
"optimizer_enable_nljoin",
"optimizer_enable_outerjoin_rewrite",
"optimizer_enable_outerjoin_to_unionall_rewrite",
+ "optimizer_enable_parallel_append",
"optimizer_enable_partition_propagation",
"optimizer_enable_partition_selection",
"optimizer_enable_push_join_below_union_all",
diff --git a/src/test/regress/GNUmakefile b/src/test/regress/GNUmakefile
index 239302227e0..f8c0637339b 100644
--- a/src/test/regress/GNUmakefile
+++ b/src/test/regress/GNUmakefile
@@ -224,7 +224,7 @@ installcheck-cbdb-parallel: all tablespace-setup twophase_pqexecparams
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'; \
+ export PGOPTIONS='-c optimizer=on -c optimizer_enable_parallel_append=true -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) \
)
diff --git a/src/test/regress/excluded_tests.conf b/src/test/regress/excluded_tests.conf
index e29197946d4..dd6b827092a 100644
--- a/src/test/regress/excluded_tests.conf
+++ b/src/test/regress/excluded_tests.conf
@@ -12,3 +12,6 @@ bfv_dd
bfv_dd_multicolumn
planhints
rowhints
+bfv_partition_plans
+qp_dpe
+partition_locking
diff --git a/src/test/regress/expected/orca_parallel_append.out b/src/test/regress/expected/orca_parallel_append.out
new file mode 100644
index 00000000000..2e24993bde5
--- /dev/null
+++ b/src/test/regress/expected/orca_parallel_append.out
@@ -0,0 +1,354 @@
+--
+-- ORCA Parallel Append
+--
+-- ORCA can implement a scan of a partitioned table as a parallel-aware
+-- Append over parallel scans of the selected partitions, instead of a
+-- serial Dynamic Seq Scan. The alternative is gated by
+-- optimizer_enable_parallel_append and competes with the dynamic scan
+-- on cost.
+--
+-- All GUCs are set locally in this file; the plans must stay stable with
+-- 3 segments and max_parallel_workers_per_gather = 2.
+--
+create schema orca_parallel_append;
+set search_path = orca_parallel_append, public;
+set optimizer = on;
+set optimizer_trace_fallback = on;
+create table pa_ao (id int, d int, pad text)
+with (appendonly = true)
+distributed by (id)
+partition by range (d) (start (1) end (6) every (1));
+insert into pa_ao
+select g, 1 + (g % 5), repeat('x', 20) from generate_series(1, 5000) g;
+analyze pa_ao;
+-- plain heap table, single level partitioning
+create table pa_heap (id int, d int)
+distributed by (id)
+partition by list (d) (partition p1 values (1), partition p2 values (2), partition p3 values (3));
+insert into pa_heap select g, 1 + (g % 3) from generate_series(1, 300) g;
+analyze pa_heap;
+-- empty partitioned table
+create table pa_empty (id int, d int)
+distributed by (id)
+partition by range (d) (start (1) end (4) every (1));
+analyze pa_empty;
+-- non-partitioned side for joins and set operations
+create table pa_dim (d int, name text) distributed by (d);
+insert into pa_dim values (1, 'one'), (2, 'two'), (3, 'three');
+analyze pa_dim;
+set enable_parallel = on;
+set max_parallel_workers_per_gather = 2;
+set parallel_setup_cost = 0;
+set min_parallel_table_scan_size = 0;
+set min_parallel_index_scan_size = 0;
+--
+-- GUC off: serial dynamic scan
+--
+set optimizer_enable_parallel_append = off;
+explain (costs off) select count(*) from pa_ao;
+ QUERY PLAN
+----------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 3:1 (slice1; segments: 3)
+ -> Partial Aggregate
+ -> Dynamic Seq Scan on pa_ao
+ Number of partitions to scan: 5 (out of 5)
+ Optimizer: GPORCA
+(6 rows)
+
+select count(*) from pa_ao;
+ count
+-------
+ 5000
+(1 row)
+
+--
+-- GUC on: parallel append over parallel seq scans
+--
+set optimizer_enable_parallel_append = on;
+explain (costs off) select count(*) from pa_ao;
+ QUERY PLAN
+------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_ao_1_prt_1
+ -> Parallel Seq Scan on pa_ao_1_prt_2
+ -> Parallel Seq Scan on pa_ao_1_prt_3
+ -> Parallel Seq Scan on pa_ao_1_prt_4
+ -> Parallel Seq Scan on pa_ao_1_prt_5
+ Optimizer: GPORCA
+(10 rows)
+
+select count(*) from pa_ao;
+ count
+-------
+ 5000
+(1 row)
+
+select d, count(*) from pa_ao group by d order by d;
+ d | count
+---+-------
+ 1 | 1000
+ 2 | 1000
+ 3 | 1000
+ 4 | 1000
+ 5 | 1000
+(5 rows)
+
+-- heap partitions
+explain (costs off) select count(*) from pa_heap;
+ QUERY PLAN
+---------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_heap_1_prt_p1
+ -> Parallel Seq Scan on pa_heap_1_prt_p2
+ -> Parallel Seq Scan on pa_heap_1_prt_p3
+ Optimizer: GPORCA
+(8 rows)
+
+select count(*) from pa_heap;
+ count
+-------
+ 300
+(1 row)
+
+-- static partition pruning: multiple, one and zero partitions survive
+explain (costs off) select count(*) from pa_ao where d >= 4;
+ QUERY PLAN
+------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_ao_1_prt_4
+ Filter: (d >= 4)
+ -> Parallel Seq Scan on pa_ao_1_prt_5
+ Filter: (d >= 4)
+ Optimizer: GPORCA
+(9 rows)
+
+select count(*) from pa_ao where d >= 4;
+ count
+-------
+ 2000
+(1 row)
+
+explain (costs off) select count(*) from pa_ao where d = 2;
+ QUERY PLAN
+------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_ao_1_prt_2
+ Filter: (d = 2)
+ Optimizer: GPORCA
+(7 rows)
+
+select count(*) from pa_ao where d = 2;
+ count
+-------
+ 1000
+(1 row)
+
+explain (costs off) select count(*) from pa_ao where d = 42;
+ QUERY PLAN
+--------------------------------
+ Aggregate
+ -> Result
+ One-Time Filter: false
+ Optimizer: GPORCA
+(4 rows)
+
+select count(*) from pa_ao where d = 42;
+ count
+-------
+ 0
+(1 row)
+
+-- empty partitioned table
+explain (costs off) select count(*) from pa_empty;
+ QUERY PLAN
+---------------------------------------------------------
+ Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_empty_1_prt_1
+ -> Parallel Seq Scan on pa_empty_1_prt_2
+ -> Parallel Seq Scan on pa_empty_1_prt_3
+ Optimizer: GPORCA
+(7 rows)
+
+select count(*) from pa_empty;
+ count
+-------
+ 0
+(1 row)
+
+-- dynamic partition elimination: the partition selector stays on the
+-- hash side, the parallel append consumes its selector id
+explain (costs off)
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+ QUERY PLAN
+---------------------------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Hash Join
+ Hash Cond: (pa_ao_1_prt_1.d = pa_dim.d)
+ -> Parallel Append
+ Partition Selectors: $0
+ -> Parallel Seq Scan on pa_ao_1_prt_1
+ -> Parallel Seq Scan on pa_ao_1_prt_2
+ -> Parallel Seq Scan on pa_ao_1_prt_3
+ -> Parallel Seq Scan on pa_ao_1_prt_4
+ -> Parallel Seq Scan on pa_ao_1_prt_5
+ -> Hash
+ -> Partition Selector (selector id: $0)
+ -> Broadcast Motion 6:6 (slice2; segments: 6)
+ -> Parallel Seq Scan on pa_dim
+ Filter: (name = 'two'::text)
+ Optimizer: GPORCA
+(18 rows)
+
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+ count
+-------
+ 1000
+(1 row)
+
+-- predicate and projection pushed into the per-partition scans
+explain (costs off) select id, d from pa_ao where id < 10 order by id;
+ QUERY PLAN
+------------------------------------------------------
+ Gather Motion 6:1 (slice1; segments: 6)
+ Merge Key: pa_ao_1_prt_1.id
+ -> Sort
+ Sort Key: pa_ao_1_prt_1.id
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_ao_1_prt_1
+ Filter: (id < 10)
+ -> Parallel Seq Scan on pa_ao_1_prt_2
+ Filter: (id < 10)
+ -> Parallel Seq Scan on pa_ao_1_prt_3
+ Filter: (id < 10)
+ -> Parallel Seq Scan on pa_ao_1_prt_4
+ Filter: (id < 10)
+ -> Parallel Seq Scan on pa_ao_1_prt_5
+ Filter: (id < 10)
+ Optimizer: GPORCA
+(16 rows)
+
+select id, d from pa_ao where id < 10 order by id;
+ id | d
+----+---
+ 1 | 2
+ 2 | 3
+ 3 | 4
+ 4 | 5
+ 5 | 1
+ 6 | 2
+ 7 | 3
+ 8 | 4
+ 9 | 5
+(9 rows)
+
+-- set operations with a partitioned child are not parallelized: a serial
+-- sibling under the same Append would be executed by every worker and
+-- its rows duplicated
+explain (costs off) select d from pa_dim union all select d from pa_heap;
+ QUERY PLAN
+----------------------------------------------------------
+ Gather Motion 3:1 (slice1; segments: 3)
+ -> Append
+ -> Seq Scan on pa_dim
+ -> Dynamic Seq Scan on pa_heap
+ Number of partitions to scan: 3 (out of 3)
+ Optimizer: GPORCA
+(6 rows)
+
+select d, count(*) from (select d from pa_dim union all select d from pa_heap) s group by d order by d;
+ d | count
+---+-------
+ 1 | 101
+ 2 | 101
+ 3 | 101
+(3 rows)
+
+select count(*) from (select d from pa_dim union select d from pa_heap) s;
+ count
+-------
+ 3
+(1 row)
+
+-- nested loop inner side requires rewindability, which the parallel append
+-- cannot provide; the plan must still be correct
+set optimizer_enable_hashjoin = off;
+explain (costs off)
+select count(*) from pa_dim join pa_heap on pa_dim.d = pa_heap.d;
+ QUERY PLAN
+---------------------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather Motion 6:1 (slice1; segments: 6)
+ -> Partial Aggregate
+ -> Nested Loop
+ Join Filter: (pa_dim.d = pa_heap_1_prt_p1.d)
+ -> Broadcast Motion 6:6 (slice2; segments: 6)
+ -> Parallel Seq Scan on pa_dim
+ -> Materialize
+ -> Parallel Append
+ -> Parallel Seq Scan on pa_heap_1_prt_p1
+ -> Parallel Seq Scan on pa_heap_1_prt_p2
+ -> Parallel Seq Scan on pa_heap_1_prt_p3
+ Optimizer: GPORCA
+(13 rows)
+
+select count(*) from pa_dim join pa_heap on pa_dim.d = pa_heap.d;
+ count
+-------
+ 300
+(1 row)
+
+reset optimizer_enable_hashjoin;
+-- results match the serial plan and the Postgres planner
+set optimizer_enable_parallel_append = off;
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+ count
+-------
+ 1000
+(1 row)
+
+set optimizer = off;
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+ count
+-------
+ 1000
+(1 row)
+
+select count(*) from pa_ao;
+ count
+-------
+ 5000
+(1 row)
+
+reset optimizer;
+reset optimizer_enable_parallel_append;
+reset enable_parallel;
+reset max_parallel_workers_per_gather;
+reset parallel_setup_cost;
+reset min_parallel_table_scan_size;
+reset min_parallel_index_scan_size;
+reset optimizer_trace_fallback;
+-- start_ignore
+drop schema orca_parallel_append cascade;
+NOTICE: drop cascades to 4 other objects
+DETAIL: drop cascades to table pa_ao
+drop cascades to table pa_heap
+drop cascades to table pa_empty
+drop cascades to table pa_dim
+-- end_ignore
diff --git a/src/test/regress/greenplum_schedule b/src/test/regress/greenplum_schedule
index 73fb0fa7a54..fac4650f7bc 100755
--- a/src/test/regress/greenplum_schedule
+++ b/src/test/regress/greenplum_schedule
@@ -297,6 +297,9 @@ test: freeze_aux_tables
# cbdb parallel test
test: cbdb_parallel
+# ORCA parallel append over partitioned tables
+test: orca_parallel_append
+
# window function parallel test
test: window_parallel
diff --git a/src/test/regress/sql/orca_parallel_append.sql b/src/test/regress/sql/orca_parallel_append.sql
new file mode 100644
index 00000000000..3d492c24f79
--- /dev/null
+++ b/src/test/regress/sql/orca_parallel_append.sql
@@ -0,0 +1,125 @@
+--
+-- ORCA Parallel Append
+--
+-- ORCA can implement a scan of a partitioned table as a parallel-aware
+-- Append over parallel scans of the selected partitions, instead of a
+-- serial Dynamic Seq Scan. The alternative is gated by
+-- optimizer_enable_parallel_append and competes with the dynamic scan
+-- on cost.
+--
+-- All GUCs are set locally in this file; the plans must stay stable with
+-- 3 segments and max_parallel_workers_per_gather = 2.
+--
+create schema orca_parallel_append;
+set search_path = orca_parallel_append, public;
+set optimizer = on;
+set optimizer_trace_fallback = on;
+
+create table pa_ao (id int, d int, pad text)
+with (appendonly = true)
+distributed by (id)
+partition by range (d) (start (1) end (6) every (1));
+
+insert into pa_ao
+select g, 1 + (g % 5), repeat('x', 20) from generate_series(1, 5000) g;
+analyze pa_ao;
+
+-- plain heap table, single level partitioning
+create table pa_heap (id int, d int)
+distributed by (id)
+partition by list (d) (partition p1 values (1), partition p2 values (2), partition p3 values (3));
+insert into pa_heap select g, 1 + (g % 3) from generate_series(1, 300) g;
+analyze pa_heap;
+
+-- empty partitioned table
+create table pa_empty (id int, d int)
+distributed by (id)
+partition by range (d) (start (1) end (4) every (1));
+analyze pa_empty;
+
+-- non-partitioned side for joins and set operations
+create table pa_dim (d int, name text) distributed by (d);
+insert into pa_dim values (1, 'one'), (2, 'two'), (3, 'three');
+analyze pa_dim;
+
+set enable_parallel = on;
+set max_parallel_workers_per_gather = 2;
+set parallel_setup_cost = 0;
+set min_parallel_table_scan_size = 0;
+set min_parallel_index_scan_size = 0;
+
+--
+-- GUC off: serial dynamic scan
+--
+set optimizer_enable_parallel_append = off;
+explain (costs off) select count(*) from pa_ao;
+select count(*) from pa_ao;
+
+--
+-- GUC on: parallel append over parallel seq scans
+--
+set optimizer_enable_parallel_append = on;
+explain (costs off) select count(*) from pa_ao;
+select count(*) from pa_ao;
+select d, count(*) from pa_ao group by d order by d;
+
+-- heap partitions
+explain (costs off) select count(*) from pa_heap;
+select count(*) from pa_heap;
+
+-- static partition pruning: multiple, one and zero partitions survive
+explain (costs off) select count(*) from pa_ao where d >= 4;
+select count(*) from pa_ao where d >= 4;
+explain (costs off) select count(*) from pa_ao where d = 2;
+select count(*) from pa_ao where d = 2;
+explain (costs off) select count(*) from pa_ao where d = 42;
+select count(*) from pa_ao where d = 42;
+
+-- empty partitioned table
+explain (costs off) select count(*) from pa_empty;
+select count(*) from pa_empty;
+
+-- dynamic partition elimination: the partition selector stays on the
+-- hash side, the parallel append consumes its selector id
+explain (costs off)
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+
+-- predicate and projection pushed into the per-partition scans
+explain (costs off) select id, d from pa_ao where id < 10 order by id;
+select id, d from pa_ao where id < 10 order by id;
+
+-- set operations with a partitioned child are not parallelized: a serial
+-- sibling under the same Append would be executed by every worker and
+-- its rows duplicated
+explain (costs off) select d from pa_dim union all select d from pa_heap;
+select d, count(*) from (select d from pa_dim union all select d from pa_heap) s group by d order by d;
+select count(*) from (select d from pa_dim union select d from pa_heap) s;
+
+-- nested loop inner side requires rewindability, which the parallel append
+-- cannot provide; the plan must still be correct
+set optimizer_enable_hashjoin = off;
+explain (costs off)
+select count(*) from pa_dim join pa_heap on pa_dim.d = pa_heap.d;
+select count(*) from pa_dim join pa_heap on pa_dim.d = pa_heap.d;
+reset optimizer_enable_hashjoin;
+
+-- results match the serial plan and the Postgres planner
+set optimizer_enable_parallel_append = off;
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+set optimizer = off;
+select count(*) from pa_ao join pa_dim using (d) where pa_dim.name = 'two';
+select count(*) from pa_ao;
+
+reset optimizer;
+reset optimizer_enable_parallel_append;
+reset enable_parallel;
+reset max_parallel_workers_per_gather;
+reset parallel_setup_cost;
+reset min_parallel_table_scan_size;
+reset min_parallel_index_scan_size;
+reset optimizer_trace_fallback;
+
+-- start_ignore
+drop schema orca_parallel_append cascade;
+-- end_ignore
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org