Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit6e9542c

Browse files
committed
Fix handling joins in VOPS FDW
1 parentc97be4e commit6e9542c

File tree

2 files changed

+94
-16
lines changed

2 files changed

+94
-16
lines changed

‎tpch2.sql‎

Lines changed: 64 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,70 @@ order by
161161
revenuedesc;
162162
-- Time: 30911.679 ms (00:30.912)
163163

164+
165+
166+
create foreign table lineitem_fdw (
167+
l_suppkey int4not null,
168+
l_orderkey int4not null,
169+
l_partkey int4not null,
170+
l_shipdatedatenot null,
171+
l_quantity float4not null,
172+
l_extendedprice float4not null,
173+
l_discount float4not null,
174+
l_tax float4not null,
175+
l_returnflag"char"not null,
176+
l_linestatus"char"not null
177+
) server vops_server options (table_name'vlineitem');
178+
179+
create foreign table orders_fdw (
180+
o_orderkey int4not null,
181+
o_custkey int4not null,
182+
o_orderstatus"char"not null,
183+
o_totalprice float4not null,
184+
o_orderdatedatenot null,
185+
o_shippriority int4not null
186+
) server vops_server options (table_name'vorders');
187+
188+
create foreign table customer_fdw (
189+
c_custkey int4not null,
190+
c_nationkey int4not null,
191+
c_acctbal float4not null
192+
) server vops_server options (table_name'vcustomer');
193+
194+
create foreign table supplier_fdw (
195+
s_suppkey int4not null,
196+
s_nationkey int4not null,
197+
s_acctbal float4not null
198+
) server vops_server options (table_name'vsupplier');
199+
200+
set enable_material=false;
201+
set enable_mergejoin=false;
202+
203+
204+
select
205+
n_name,
206+
count(*),
207+
sum(l_extendedprice* (1-l_discount))as revenue
208+
from
209+
customer_fdwjoin orders_fdwon c_custkey= o_custkey
210+
join lineitem_fdwon l_orderkey= o_orderkey
211+
join supplier_fdwon l_suppkey= s_suppkey
212+
join nationon c_nationkey= n_nationkey
213+
join regionon n_regionkey= r_regionkey
214+
where
215+
c_nationkey= s_nationkey
216+
and r_name='ASIA'
217+
and o_orderdate>='1996-01-01'
218+
and o_orderdate<'1997-01-01'
219+
group by
220+
n_name
221+
order by
222+
revenuedesc;
223+
-- Time: 55101.292 ms (00:55.101)
224+
225+
226+
\echo Foreign data wrapper result
227+
164228
createtablehlineitem(
165229
l_suppkey int4not null,
166230
l_orderkey int4not null,
@@ -202,10 +266,6 @@ create function unnest_supplier(vsupplier) returns setof hsupplier as 'vops','vo
202266
createfunctionunnest_lineitem(vlineitem) returns setof hlineitemas'vops','vops_unnest' language C parallel safe immutable strict;
203267
createfunctionunnest_orders(vorders) returns setof hordersas'vops','vops_unnest' language C parallel safe immutable strict;
204268

205-
set enable_material=false;
206-
set enable_mergejoin=false;
207-
208-
209269
select
210270
n_name,
211271
count(*),

‎vops_fdw.c‎

Lines changed: 30 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,8 @@ typedef struct PgFdwScanState
8181
Portalportal;/* SPI portal */
8282
intnumParams;/* number of parameters passed to query */
8383
inttile_pos;
84+
uint64table_pos;
8485
HeapTuplespi_tuple;
85-
SPITupleTable*spi_tuptable;
8686
Datum*src_values;
8787
Datum*dst_values;
8888
bool*src_nulls;
@@ -629,9 +629,13 @@ postgresIterateForeignScan(ForeignScanState *node)
629629
MemoryContextoldcontext=MemoryContextSwitchTo(fsstate->spi_context);
630630

631631
while (true) {
632-
if (fsstate->spi_tuptable==NULL) {
633-
SPI_cursor_fetch(fsstate->portal, true,1);
634-
if (!SPI_processed) {
632+
if (fsstate->spi_tuple==NULL) {
633+
if (fsstate->portal!=NULL) {
634+
filter_mask= ~0;
635+
SPI_cursor_fetch(fsstate->portal, true,1);
636+
fsstate->table_pos=0;
637+
}
638+
if (fsstate->table_pos==SPI_processed) {
635639
MemoryContextSwitchTo(oldcontext);
636640
returnExecClearTuple(slot);
637641
}
@@ -642,8 +646,7 @@ postgresIterateForeignScan(ForeignScanState *node)
642646
fsstate->tile_pos=0;
643647
fsstate->filter_mask=filter_mask;
644648
}
645-
fsstate->spi_tuptable=SPI_tuptable;
646-
fsstate->spi_tuple=SPI_tuptable->vals[0];
649+
fsstate->spi_tuple=SPI_tuptable->vals[fsstate->table_pos++];
647650
if (fsstate->vops_types==NULL) {
648651
fsstate->vops_types=palloc(sizeof(vops_type_info)*n_attrs);
649652
fsstate->attr_types=palloc(sizeof(Oid)*n_attrs);
@@ -672,7 +675,7 @@ postgresIterateForeignScan(ForeignScanState *node)
672675
if (i>0)
673676
{
674677
/* ordinary column */
675-
fsstate->src_values[i-1]=SPI_getbinval(fsstate->spi_tuple,fsstate->spi_tuptable->tupdesc,j+1,&fsstate->src_nulls[i-1]);
678+
fsstate->src_values[i-1]=SPI_getbinval(fsstate->spi_tuple,SPI_tuptable->tupdesc,j+1,&fsstate->src_nulls[i-1]);
676679
}
677680
j+=1;
678681
}
@@ -742,8 +745,10 @@ postgresIterateForeignScan(ForeignScanState *node)
742745
NextTuple:;
743746
}
744747
SPI_freetuple(fsstate->spi_tuple);
745-
SPI_freetuptable(fsstate->spi_tuptable);
746-
fsstate->spi_tuptable=NULL;
748+
if (fsstate->portal) {
749+
SPI_freetuptable(SPI_tuptable);
750+
}
751+
fsstate->spi_tuple=NULL;
747752
}
748753
}
749754

@@ -760,6 +765,8 @@ postgresReScanForeignScan(ForeignScanState *node)
760765
bool*nulls=NULL;
761766
MemoryContextoldcontext=MemoryContextSwitchTo(fsstate->spi_context);
762767
Oid*argtypes=NULL;
768+
intrc;
769+
763770
if (fsstate->numParams>0) {
764771
ExprContext*econtext=node->ss.ps.ps_ExprContext;
765772
ForeignScan*fsplan= (ForeignScan*)node->ss.ps.plan;
@@ -780,8 +787,17 @@ postgresReScanForeignScan(ForeignScanState *node)
780787
i+=1;
781788
}
782789
}
783-
fsstate->portal=SPI_cursor_open_with_args(NULL,fsstate->query,fsstate->numParams,argtypes,values,nulls, true,CURSOR_OPT_PARALLEL_OK);
784-
fsstate->spi_tuptable=NULL;
790+
if (fsstate->rel==NULL) {/* aggregate is pushed down: do not use cusror to allow parallel query execution */
791+
rc=SPI_execute_with_args(fsstate->query,fsstate->numParams,argtypes,values,nulls, true,0);
792+
if (rc!=SPI_OK_SELECT) {
793+
elog(ERROR,"Failed to execute VOPS query %s: %d",fsstate->query,rc);
794+
}
795+
fsstate->portal=NULL;
796+
}else {
797+
fsstate->portal=SPI_cursor_open_with_args(NULL,fsstate->query,fsstate->numParams,argtypes,values,nulls, true,CURSOR_OPT_PARALLEL_OK);
798+
}
799+
fsstate->table_pos=0;
800+
fsstate->spi_tuple=NULL;
785801

786802
MemoryContextSwitchTo(oldcontext);
787803
}
@@ -800,7 +816,9 @@ postgresEndForeignScan(ForeignScanState *node)
800816
{
801817
MemoryContextoldcontext=MemoryContextSwitchTo(fsstate->spi_context);
802818

803-
SPI_cursor_close(fsstate->portal);
819+
if (fsstate->portal) {
820+
SPI_cursor_close(fsstate->portal);
821+
}
804822
SPI_finish();
805823

806824
MemoryContextSwitchTo(oldcontext);

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp