Skip to content

Commit 5371547

Browse files
knizhnikknizhnik
committed
VOPS as FDW
1 parent df44e1a commit 5371547

File tree

4 files changed

+103
-139
lines changed

4 files changed

+103
-139
lines changed

‎Makefile

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
# contrib/vops/Makefile
22

33
MODULE_big = vops
4-
OBJS = vops.o
4+
OBJS = vops.o vops_fdw.o deparse.o
55

66
EXTENSION = vops
77
DATA = vops--1.0.sql
88
PGFILEDESC = "vops - vectorized operations"
9-
CUSTOM_COPT = -O3
9+
CUSTOM_COPT = -O0
1010

1111
REGRESS = test
1212

‎tpch.sql

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -207,3 +207,79 @@ order by
207207
l_linestatus;
208208
-- Seq time: 1490.143 ms
209209
-- Par time: 396.329 ms
210+
211+
212+
-- Access through FDW
213+
214+
create foreign table lineitem_fdw (
215+
l_shipdate date not null,
216+
l_quantity float4 not null,
217+
l_extendedprice float4 not null,
218+
l_discount float4 not null,
219+
l_tax float4 not null,
220+
l_returnflag "char" not null,
221+
l_linestatus "char" not null
222+
) server vops_server options (table_name 'vops_lineitem');
223+
224+
select
225+
l_returnflag,
226+
l_linestatus,
227+
sum(l_quantity) as sum_qty,
228+
sum(l_extendedprice) as sum_base_price,
229+
sum(l_extendedprice*(1-l_discount)) as sum_disc_price,
230+
sum(l_extendedprice*(1-l_discount)*(1+l_tax)) as sum_charge,
231+
avg(l_quantity) as avg_qty,
232+
avg(l_extendedprice) as avg_price,
233+
avg(l_discount) as avg_disc,
234+
count(*) as count_order
235+
from
236+
lineitem_fdw
237+
where
238+
l_shipdate <= '1998-12-01'
239+
group by
240+
l_returnflag,
241+
l_linestatus
242+
order by
243+
l_returnflag,
244+
l_linestatus;
245+
246+
select
247+
sum(l_extendedprice*l_discount) as revenue
248+
from
249+
lineitem_fdw
250+
where
251+
l_shipdate between '1996-01-01' and '1997-01-01'
252+
and l_discount between 0.08 and 0.1
253+
and l_quantity < 24;
254+
255+
create foreign table lineitem_projection_fdw (
256+
l_shipdate date not null,
257+
l_quantity float4 not null,
258+
l_extendedprice float4 not null,
259+
l_discount float4 not null,
260+
l_tax float4 not null,
261+
l_returnflag "char" not null,
262+
l_linestatus "char" not null
263+
) server vops_server options (table_name 'vops_lineitem_projection');
264+
265+
select
266+
l_returnflag,
267+
l_linestatus,
268+
sum(l_quantity) as sum_qty,
269+
sum(l_extendedprice) as sum_base_price,
270+
sum(l_extendedprice*(1-l_discount)) as sum_disc_price,
271+
sum(l_extendedprice*(1-l_discount)*(1+l_tax)) as sum_charge,
272+
avg(l_quantity) as avg_qty,
273+
avg(l_extendedprice) as avg_price,
274+
avg(l_discount) as avg_disc,
275+
count(*) as count_order
276+
from
277+
lineitem_projection_fdw
278+
where
279+
l_shipdate <= '1998-12-01'
280+
group by
281+
l_returnflag,
282+
l_linestatus
283+
order by
284+
l_returnflag,
285+
l_linestatus;

‎vops--1.0.sql

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2651,3 +2651,19 @@ create cast (vops_bool as bool) with function filter(vops_bool) AS IMPLICIT;
26512651
create function is_null(anyelement) returns vops_bool as 'MODULE_PATHNAME','vops_is_null' language C parallel safe immutable;
26522652
create function is_not_null(anyelement) returns vops_bool as 'MODULE_PATHNAME','vops_is_not_null' language C parallel safe immutable;
26532653

2654+
CREATE FUNCTION vops_fdw_handler()
2655+
RETURNS fdw_handler
2656+
AS 'MODULE_PATHNAME'
2657+
LANGUAGE C STRICT;
2658+
2659+
CREATE FUNCTION vops_fdw_validator(text[], oid)
2660+
RETURNS void
2661+
AS 'MODULE_PATHNAME'
2662+
LANGUAGE C STRICT;
2663+
2664+
CREATE FOREIGN DATA WRAPPER vops_fdw
2665+
HANDLER vops_fdw_handler
2666+
VALIDATOR vops_fdw_validator;
2667+
2668+
CREATE SERVER vops_server FOREIGN DATA WRAPPER vops_fdw;
2669+

‎vops.c

Lines changed: 9 additions & 137 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
#include "nodes/nodeFuncs.h"
4444
#include "nodes/makefuncs.h"
4545
#include "nodes/pg_list.h"
46+
#include "vops.h"
4647

4748
#ifdef PG_MODULE_MAGIC
4849
PG_MODULE_MAGIC;
@@ -52,142 +53,7 @@ PG_MODULE_MAGIC;
5253
void _PG_init(void);
5354
void _PG_fini(void);
5455

55-
typedef enum
56-
{
57-
VOPS_BOOL,
58-
VOPS_CHAR,
59-
VOPS_INT2,
60-
VOPS_INT4,
61-
VOPS_INT8,
62-
VOPS_DATE,
63-
VOPS_TIMESTAMP,
64-
VOPS_FLOAT4,
65-
VOPS_FLOAT8,
66-
VOPS_LAST,
67-
} vops_type;
68-
69-
typedef enum
70-
{
71-
VOPS_AGG_SUM,
72-
VOPS_AGG_AVG,
73-
VOPS_AGG_MAX,
74-
VOPS_AGG_MIN,
75-
VOPS_AGG_COUNT,
76-
VOPS_AGG_LAST
77-
} vops_agg_kind;
78-
79-
80-
#define TILE_SIZE 64 /* just because of maximum size of bitmask */
81-
#define MAX_SQL_STMT_LEN 1024
82-
#define MAX_CSV_LINE_LEN 4096
83-
#define MAX_TILE_STRLEN (TILE_SIZE*16)
84-
#define INIT_MAP_SIZE (1024*1024)
85-
86-
static uint64 filter_mask;
87-
typedef long long long64;
88-
89-
/* Common prefix for all tile */
90-
typedef struct {
91-
uint64 null_mask;
92-
uint64 empty_mask;
93-
} vops_tile_hdr;
94-
95-
#define TILE(TYPE,CTYPE) \
96-
typedef struct { \
97-
vops_tile_hdr hdr; \
98-
CTYPE payload[TILE_SIZE]; \
99-
} vops_##TYPE
100-
101-
TILE(char,char);
102-
TILE(int2,int16);
103-
TILE(int4,int32);
104-
TILE(int8,int64);
105-
TILE(float4,float4);
106-
TILE(float8,float8);
107-
108-
typedef struct {
109-
vops_tile_hdr hdr;
110-
uint64 payload;
111-
} vops_bool;
112-
113-
typedef struct {
114-
uint64 count;
115-
double sum;
116-
} vops_avg_state;
117-
118-
typedef struct {
119-
uint64 count;
120-
double sum;
121-
double sum2;
122-
} vops_var_state;
123-
124-
typedef struct {
125-
HTAB* htab;
126-
int n_aggs;
127-
vops_type agg_type;
128-
vops_agg_kind* agg_kinds;
129-
} vops_agg_state;
130-
131-
typedef union {
132-
bool b;
133-
char ch;
134-
int16 i2;
135-
int32 i4;
136-
int64 i8;
137-
float f4;
138-
double f8;
139-
} vops_value;
140-
141-
typedef struct {
142-
vops_value acc;
143-
uint64 count;
144-
} vops_agg_value;
145-
146-
typedef struct {
147-
int64 group_by;
148-
uint64 count;
149-
vops_agg_value values[1];
150-
} vops_group_by_entry;
151-
152-
#define VOPS_AGGREGATES_ATTRIBUTES 3
153-
154-
typedef struct {
155-
HASH_SEQ_STATUS iter;
156-
TupleDesc desc;
157-
Datum* elems;
158-
bool* nulls;
159-
int16 elmlen;
160-
bool elmbyval;
161-
char elmalign;
162-
} vops_reduce_context;
163-
164-
typedef struct {
165-
Datum* values;
166-
bool* nulls;
167-
vops_type* types;
168-
TupleDesc desc;
169-
int n_attrs;
170-
int tile_pos;
171-
uint64 filter_mask;
172-
vops_tile_hdr** tiles;
173-
} vops_unnest_context;
174-
175-
typedef struct {
176-
vops_float8 tile;
177-
double sum;
178-
double sum2;
179-
uint64 count;
180-
} vops_window_state;
181-
182-
183-
typedef struct {
184-
vops_type tid;
185-
int16 len;
186-
bool byval;
187-
char align;
188-
FmgrInfo inproc;
189-
Oid inproc_param_oid;
190-
} vops_type_info;
56+
uint64 filter_mask;
19157

19258
static struct {
19359
char const* name;
@@ -233,7 +99,7 @@ static vops_agg_state* vops_init_agg_state(char const* aggregates, Oid elem_type
23399
static vops_agg_state* vops_create_agg_state(int n_aggregates);
234100
static void vops_agg_state_accumulate(vops_agg_state* state, int64 group_by, int i, Datum* tiles, bool* nulls);
235101

236-
static vops_type vops_get_type(Oid typid)
102+
vops_type vops_get_type(Oid typid)
237103
{
238104
int i;
239105
if (vops_type_map[0].oid == InvalidOid) {
@@ -1587,6 +1453,12 @@ Datum vops_populate(PG_FUNCTION_ARGS)
15871453
HeapTuple spi_tuple = SPI_tuptable->vals[0];
15881454
spi_tupdesc = SPI_tuptable->tupdesc;
15891455
if (j == TILE_SIZE) {
1456+
for (i = 0; i < n_attrs; i++) {
1457+
if (types[i].tid != VOPS_LAST) {
1458+
vops_tile_hdr* tile = (vops_tile_hdr*)DatumGetPointer(values[i]);
1459+
tile->empty_mask = 0;
1460+
}
1461+
}
15901462
insert_tuple(values, nulls);
15911463
j = 0;
15921464
}

0 commit comments

Comments
 (0)