public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 08/20] union{} with a CompressionAlgorithm alg
3+ messages / 3 participants
[nested] [flat]
* [PATCH 08/20] union{} with a CompressionAlgorithm alg
@ 2020-12-21 06:11 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Justin Pryzby @ 2020-12-21 06:11 UTC (permalink / raw)
---
src/bin/pg_dump/compress_io.c | 200 ++++++++++++++++++----------------
src/bin/pg_dump/pg_dump.c | 2 +-
2 files changed, 106 insertions(+), 96 deletions(-)
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index fa94148cdf..e07436bc21 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -651,23 +651,27 @@ WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs,
*/
struct cfp
{
- FILE *uncompressedfp;
+ CompressionAlgorithm alg;
+
+ union {
+ FILE *fp;
+
#ifdef HAVE_LIBZ
- gzFile compressedfp;
+ gzFile gzfp;
#endif
-#ifdef HAVE_LIBZSTD // XXX: this should be a union with a CompressionAlgorithm alg?
- /* This is a normal file to which we read/write compressed data */
- struct {
- FILE *fp;
- // XXX: use one separate ZSTD_CStream per thread: disable on windows ?
- ZSTD_CStream *cstream;
- ZSTD_DStream *dstream;
- ZSTD_outBuffer output;
- ZSTD_inBuffer input;
- } zstd;
+#ifdef HAVE_LIBZSTD
+ struct {
+ /* This is a normal file to which we read/write compressed data */
+ FILE *fp;
+ // XXX: use one separate ZSTD_CStream per thread: disable on windows ?
+ ZSTD_CStream *cstream;
+ ZSTD_DStream *dstream;
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ } zstd;
#endif
-
+ } u;
};
static int hasSuffix(const char *filename);
@@ -754,6 +758,8 @@ cfopen(const char *path, const char *mode, Compress *compression)
{
cfp *fp = pg_malloc0(sizeof(cfp));
+ fp->alg = compression->alg;
+
switch (compression->alg)
{
#ifdef HAVE_LIBZ
@@ -765,15 +771,15 @@ cfopen(const char *path, const char *mode, Compress *compression)
snprintf(mode_compression, sizeof(mode_compression), "%s%d",
mode, compression->level);
- fp->compressedfp = gzopen(path, mode_compression);
+ fp->u.gzfp = gzopen(path, mode_compression);
}
else
{
/* don't specify a level, just use the zlib default */
- fp->compressedfp = gzopen(path, mode);
+ fp->u.gzfp = gzopen(path, mode);
}
- if (fp->compressedfp == NULL)
+ if (fp->u.gzfp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -783,8 +789,8 @@ cfopen(const char *path, const char *mode, Compress *compression)
#ifdef HAVE_LIBZSTD
case COMPR_ALG_ZSTD:
- fp->zstd.fp = fopen(path, mode);
- if (fp->zstd.fp == NULL)
+ fp->u.zstd.fp = fopen(path, mode);
+ if (fp->u.zstd.fp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -792,23 +798,23 @@ cfopen(const char *path, const char *mode, Compress *compression)
else if (mode[0] == 'w' || mode[0] == 'a' ||
strchr(mode, '+') != NULL)
{
- fp->zstd.output.size = ZSTD_CStreamOutSize();
- fp->zstd.output.dst = pg_malloc0(fp->zstd.output.size);
- fp->zstd.cstream = ZstdCStreamParams(compression);
+ fp->u.zstd.output.size = ZSTD_CStreamOutSize();
+ fp->u.zstd.output.dst = pg_malloc0(fp->u.zstd.output.size);
+ fp->u.zstd.cstream = ZstdCStreamParams(compression);
}
else if (strchr(mode, 'r'))
{
- fp->zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
- fp->zstd.dstream = ZSTD_createDStream();
- if (fp->zstd.dstream == NULL)
+ fp->u.zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
+ fp->u.zstd.dstream = ZSTD_createDStream();
+ if (fp->u.zstd.dstream == NULL)
fatal("could not initialize compression library");
} // XXX else: bad mode
return fp;
#endif
case COMPR_ALG_NONE:
- fp->uncompressedfp = fopen(path, mode);
- if (fp->uncompressedfp == NULL)
+ fp->u.fp = fopen(path, mode);
+ if (fp->u.fp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -830,6 +836,8 @@ cfdopen(int fd, const char *mode, Compress *compression)
{
cfp *fp = pg_malloc0(sizeof(cfp));
+ fp->alg = compression->alg;
+
switch (compression->alg)
{
#ifdef HAVE_LIBZ
@@ -841,15 +849,15 @@ cfdopen(int fd, const char *mode, Compress *compression)
snprintf(mode_compression, sizeof(mode_compression), "%s%d",
mode, compression->level);
- fp->compressedfp = gzdopen(fd, mode_compression);
+ fp->u.gzfp = gzdopen(fd, mode_compression);
}
else
{
/* don't specify a level, just use the zlib default */
- fp->compressedfp = gzdopen(fd, mode);
+ fp->u.gzfp = gzdopen(fd, mode);
}
- if (fp->compressedfp == NULL)
+ if (fp->u.gzfp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -859,8 +867,8 @@ cfdopen(int fd, const char *mode, Compress *compression)
#ifdef HAVE_LIBZSTD
case COMPR_ALG_ZSTD:
- fp->zstd.fp = fdopen(fd, mode);
- if (fp->zstd.fp == NULL)
+ fp->u.zstd.fp = fdopen(fd, mode);
+ if (fp->u.zstd.fp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -868,23 +876,23 @@ cfdopen(int fd, const char *mode, Compress *compression)
else if (mode[0] == 'w' || mode[0] == 'a' ||
strchr(mode, '+') != NULL)
{
- fp->zstd.output.size = ZSTD_CStreamOutSize();
- fp->zstd.output.dst = pg_malloc0(fp->zstd.output.size);
- fp->zstd.cstream = ZstdCStreamParams(compression);
+ fp->u.zstd.output.size = ZSTD_CStreamOutSize();
+ fp->u.zstd.output.dst = pg_malloc0(fp->u.zstd.output.size);
+ fp->u.zstd.cstream = ZstdCStreamParams(compression);
}
else if (strchr(mode, 'r'))
{
- fp->zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
- fp->zstd.dstream = ZSTD_createDStream();
- if (fp->zstd.dstream == NULL)
+ fp->u.zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
+ fp->u.zstd.dstream = ZSTD_createDStream();
+ if (fp->u.zstd.dstream == NULL)
fatal("could not initialize compression library");
} // XXX else: bad mode
return fp;
#endif
case COMPR_ALG_NONE:
- fp->uncompressedfp = fdopen(fd, mode);
- if (fp->uncompressedfp == NULL)
+ fp->u.fp = fdopen(fd, mode);
+ if (fp->u.fp == NULL)
{
free_keep_errno(fp);
fp = NULL;
@@ -908,13 +916,13 @@ cfread(void *ptr, int size, cfp *fp)
return 0;
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
+ if (fp->alg == COMPR_ALG_LIBZ)
{
- ret = gzread(fp->compressedfp, ptr, size);
- if (ret != size && !gzeof(fp->compressedfp))
+ ret = gzread(fp->u.gzfp, ptr, size);
+ if (ret != size && !gzeof(fp->u.gzfp))
{
int errnum;
- const char *errmsg = gzerror(fp->compressedfp, &errnum);
+ const char *errmsg = gzerror(fp->u.gzfp, &errnum);
fatal("could not read from input file: %s",
errnum == Z_ERRNO ? strerror(errno) : errmsg);
@@ -924,10 +932,10 @@ cfread(void *ptr, int size, cfp *fp)
#endif
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
+ if (fp->alg == COMPR_ALG_ZSTD)
{
- ZSTD_outBuffer *output = &fp->zstd.output;
- ZSTD_inBuffer *input = &fp->zstd.input;
+ ZSTD_outBuffer *output = &fp->u.zstd.output;
+ ZSTD_inBuffer *input = &fp->u.zstd.input;
size_t input_size = ZSTD_DStreamInSize();
/* input_size is the allocated size */
size_t res, cnt;
@@ -953,7 +961,7 @@ cfread(void *ptr, int size, cfp *fp)
/* read compressed data if we must produce more input */
if (input->pos == input->size)
{
- cnt = fread(unconstify(void *, input->src), 1, input_size, fp->zstd.fp);
+ cnt = fread(unconstify(void *, input->src), 1, input_size, fp->u.zstd.fp);
input->size = cnt;
/* If we have no input to consume, we're done */
@@ -968,7 +976,7 @@ cfread(void *ptr, int size, cfp *fp)
for ( ; input->pos < input->size; )
{
/* decompress */
- res = ZSTD_decompressStream(fp->zstd.dstream, output, input);
+ res = ZSTD_decompressStream(fp->u.zstd.dstream, output, input);
if (res == 0)
break; /* End of frame */
if (output->pos == output->size)
@@ -985,9 +993,9 @@ cfread(void *ptr, int size, cfp *fp)
}
#endif
- ret = fread(ptr, 1, size, fp->uncompressedfp);
- if (ret != size && !feof(fp->uncompressedfp))
- READ_ERROR_EXIT(fp->uncompressedfp);
+ ret = fread(ptr, 1, size, fp->u.fp);
+ if (ret != size && !feof(fp->u.fp))
+ READ_ERROR_EXIT(fp->u.fp);
return ret;
}
@@ -995,16 +1003,16 @@ int
cfwrite(const void *ptr, int size, cfp *fp)
{
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
- return gzwrite(fp->compressedfp, ptr, size);
+ if (fp->alg == COMPR_ALG_LIBZ)
+ return gzwrite(fp->u.gzfp, ptr, size);
#endif
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
+ if (fp->alg == COMPR_ALG_ZSTD)
{
size_t res, cnt;
- ZSTD_outBuffer *output = &fp->zstd.output;
- ZSTD_inBuffer *input = &fp->zstd.input;
+ ZSTD_outBuffer *output = &fp->u.zstd.output;
+ ZSTD_inBuffer *input = &fp->u.zstd.input;
input->src = ptr;
input->size = size;
@@ -1014,11 +1022,11 @@ cfwrite(const void *ptr, int size, cfp *fp)
while (input->pos != input->size)
{
output->pos = 0;
- res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_continue);
+ res = ZSTD_compressStream2(fp->u.zstd.cstream, output, input, ZSTD_e_continue);
if (ZSTD_isError(res))
fatal("could not compress data: %s", ZSTD_getErrorName(res));
- cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ cnt = fwrite(output->dst, 1, output->pos, fp->u.zstd.fp);
if (cnt != output->pos)
fatal("could not write data: %s", strerror(errno));
}
@@ -1027,7 +1035,7 @@ cfwrite(const void *ptr, int size, cfp *fp)
}
#endif
- return fwrite(ptr, 1, size, fp->uncompressedfp);
+ return fwrite(ptr, 1, size, fp->u.fp);
}
int
@@ -1036,12 +1044,12 @@ cfgetc(cfp *fp)
int ret;
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
+ if (fp->alg == COMPR_ALG_LIBZ)
{
- ret = gzgetc(fp->compressedfp);
+ ret = gzgetc(fp->u.gzfp);
if (ret == EOF)
{
- if (!gzeof(fp->compressedfp))
+ if (!gzeof(fp->u.gzfp))
fatal("could not read from input file: %s", strerror(errno));
else
fatal("could not read from input file: end of file");
@@ -1051,11 +1059,11 @@ cfgetc(cfp *fp)
#endif
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
+ if (fp->alg == COMPR_ALG_ZSTD)
{
if (cfread(&ret, 1, fp) != 1)
{
- if (feof(fp->zstd.fp))
+ if (feof(fp->u.zstd.fp))
fatal("could not read from input file: end of file");
else
fatal("could not read from input file: %s", strerror(errno));
@@ -1064,9 +1072,9 @@ cfgetc(cfp *fp)
}
#endif
- ret = fgetc(fp->uncompressedfp);
+ ret = fgetc(fp->u.fp);
if (ret == EOF)
- READ_ERROR_EXIT(fp->uncompressedfp);
+ READ_ERROR_EXIT(fp->u.fp);
return ret;
}
@@ -1074,11 +1082,12 @@ char *
cfgets(cfp *fp, char *buf, int len)
{
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
- return gzgets(fp->compressedfp, buf, len);
+ if (fp->alg == COMPR_ALG_LIBZ)
+ return gzgets(fp->u.gzfp, buf, len);
#endif
+
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
+ if (fp->alg == COMPR_ALG_ZSTD)
{
/*
* Read one byte at a time until newline or EOF.
@@ -1102,7 +1111,7 @@ cfgets(cfp *fp, char *buf, int len)
}
#endif
- return fgets(buf, len, fp->uncompressedfp);
+ return fgets(buf, len, fp->u.fp);
}
/* Close the given compressed or uncompressed stream; return 0 on success. */
@@ -1117,54 +1126,54 @@ cfclose(cfp *fp)
return EOF;
}
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
+ if (fp->alg == COMPR_ALG_LIBZ)
{
- result = gzclose(fp->compressedfp);
- fp->compressedfp = NULL;
+ result = gzclose(fp->u.gzfp);
+ fp->u.gzfp = NULL;
return result;
}
#endif
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
+ if (fp->alg == COMPR_ALG_ZSTD)
{
- ZSTD_outBuffer *output = &fp->zstd.output;
- ZSTD_inBuffer *input = &fp->zstd.input;
+ ZSTD_outBuffer *output = &fp->u.zstd.output;
+ ZSTD_inBuffer *input = &fp->u.zstd.input;
size_t res, cnt;
- if (fp->zstd.cstream)
+ if (fp->u.zstd.cstream)
{
for (;;)
{
output->pos = 0;
- res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_end);
+ res = ZSTD_compressStream2(fp->u.zstd.cstream, output, input, ZSTD_e_end);
if (ZSTD_isError(res))
fatal("could not compress data: %s", ZSTD_getErrorName(res));
- cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ cnt = fwrite(output->dst, 1, output->pos, fp->u.zstd.fp);
if (cnt != output->pos)
fatal("could not write data: %s", strerror(errno));
if (res == 0)
break;
}
- ZSTD_freeCStream(fp->zstd.cstream);
- pg_free(fp->zstd.output.dst);
+ ZSTD_freeCStream(fp->u.zstd.cstream);
+ pg_free(fp->u.zstd.output.dst);
}
- if (fp->zstd.dstream)
+ if (fp->u.zstd.dstream)
{
- ZSTD_freeDStream(fp->zstd.dstream);
- pg_free(unconstify(void *, fp->zstd.input.src));
+ ZSTD_freeDStream(fp->u.zstd.dstream);
+ pg_free(unconstify(void *, fp->u.zstd.input.src));
}
- result = fclose(fp->zstd.fp);
- fp->zstd.fp = NULL;
+ result = fclose(fp->u.zstd.fp);
+ fp->u.zstd.fp = NULL;
return result;
}
#endif
- result = fclose(fp->uncompressedfp);
- fp->uncompressedfp = NULL;
+ result = fclose(fp->u.fp);
+ fp->u.fp = NULL;
free_keep_errno(fp);
return result;
}
@@ -1173,25 +1182,26 @@ int
cfeof(cfp *fp)
{
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
- return gzeof(fp->compressedfp);
+ if (fp->alg == COMPR_ALG_LIBZ)
+ return gzeof(fp->u.gzfp);
#endif
#ifdef HAVE_LIBZSTD
- if (fp->zstd.fp)
- return feof(fp->zstd.fp);
+ if (fp->alg == COMPR_ALG_ZSTD)
+ return feof(fp->u.zstd.fp);
#endif
- return feof(fp->uncompressedfp);
+
+ return feof(fp->u.fp);
}
const char *
get_cfp_error(cfp *fp)
{
#ifdef HAVE_LIBZ
- if (fp->compressedfp)
+ if (fp->alg == COMPR_ALG_LIBZ)
{
int errnum;
- const char *errmsg = gzerror(fp->compressedfp, &errnum);
+ const char *errmsg = gzerror(fp->u.gzfp, &errnum);
if (errnum != Z_ERRNO)
return errmsg;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 7c2f7a9ca3..5e009e5854 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -397,7 +397,7 @@ parse_compression(const char *optarg, Compress *compress)
const int default_compress_level[] = {
0, /* COMPR_ALG_NONE */
Z_DEFAULT_COMPRESSION, /* COMPR_ALG_ZLIB */
- 0, // XXX: ZSTD_CLEVEL_DEFAULT, /* COMPR_ALG_ZSTD */
+ 0, // #ifdef LIBZSTD ZSTD_CLEVEL_DEFAULT, /* COMPR_ALG_ZSTD */
};
compress->level = default_compress_level[compress->alg];
--
2.17.0
--GRPZ8SYKNexpdSJ7
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0009-Move-zlib-into-the-union.patch"
^ permalink raw reply [nested|flat] 3+ messages in thread
* Should consider materializing the cheapest inner path in consider_parallel_nestloop()
@ 2023-09-05 08:52 tender wang <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: tender wang @ 2023-09-05 08:52 UTC (permalink / raw)
To: [email protected]
Hi all,
I recently run benchmark[1] on master, but I found performance problem
as below:
explain analyze select
subq_0.c0 as c0,
subq_0.c1 as c1,
subq_0.c2 as c2
from
(select
ref_0.l_shipmode as c0,
sample_0.l_orderkey as c1,
sample_0.l_quantity as c2,
ref_0.l_orderkey as c3,
sample_0.l_shipmode as c5,
ref_0.l_shipinstruct as c6
from
public.lineitem as ref_0
left join public.lineitem as sample_0
on ((select p_partkey from public.part order by p_partkey limit 1)
is not NULL)
where sample_0.l_orderkey is NULL) as subq_0
where subq_0.c5 is NULL
limit 1;
QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------
Limit (cost=78.00..45267050.75 rows=1 width=27) (actual
time=299695.097..299695.099 rows=0 loops=1)
InitPlan 1 (returns $0)
-> Limit (cost=78.00..78.00 rows=1 width=8) (actual
time=0.651..0.652 rows=1 loops=1)
-> Sort (cost=78.00..83.00 rows=2000 width=8) (actual
time=0.650..0.651 rows=1 loops=1)
Sort Key: part.p_partkey
Sort Method: top-N heapsort Memory: 25kB
-> Seq Scan on part (cost=0.00..68.00 rows=2000 width=8)
(actual time=0.013..0.428 rows=2000 loops=1)
-> Nested Loop Left Join (cost=0.00..45266972.75 rows=1 width=27)
(actual time=299695.096..299695.096 rows=0 loops=1)
Join Filter: ($0 IS NOT NULL)
Filter: ((sample_0.l_orderkey IS NULL) AND (sample_0.l_shipmode IS
NULL))
Rows Removed by Filter: 3621030625
-> Seq Scan on lineitem ref_0 (cost=0.00..1969.75 rows=60175
width=11) (actual time=0.026..6.225 rows=60175 loops=1)
-> Materialize (cost=0.00..2270.62 rows=60175 width=27) (actual
time=0.000..2.554 rows=60175 loops=60175)
-> Seq Scan on lineitem sample_0 (cost=0.00..1969.75
rows=60175 width=27) (actual time=0.004..8.169 rows=60175 loops=1)
Planning Time: 0.172 ms
Execution Time: 299695.501 ms
(16 rows)
After I set enable_material to off, the same query run faster, as below:
set enable_material = off;
explain analyze select
subq_0.c0 as c0,
subq_0.c1 as c1,
subq_0.c2 as c2
from
(select
ref_0.l_shipmode as c0,
sample_0.l_orderkey as c1,
sample_0.l_quantity as c2,
ref_0.l_orderkey as c3,
sample_0.l_shipmode as c5,
ref_0.l_shipinstruct as c6
from
public.lineitem as ref_0
left join public.lineitem as sample_0
on ((select p_partkey from public.part order by p_partkey limit 1)
is not NULL)
where sample_0.l_orderkey is NULL) as subq_0
where subq_0.c5 is NULL
limit 1;
QUERY
PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------
Limit (cost=1078.00..91026185.57 rows=1 width=27) (actual
time=192669.605..192670.425 rows=0 loops=1)
InitPlan 1 (returns $0)
-> Limit (cost=78.00..78.00 rows=1 width=8) (actual
time=0.662..0.663 rows=1 loops=1)
-> Sort (cost=78.00..83.00 rows=2000 width=8) (actual
time=0.661..0.662 rows=1 loops=1)
Sort Key: part.p_partkey
Sort Method: top-N heapsort Memory: 25kB
-> Seq Scan on part (cost=0.00..68.00 rows=2000 width=8)
(actual time=0.017..0.430 rows=2000 loops=1)
-> Gather (cost=1000.00..91026107.57 rows=1 width=27) (actual
time=192669.604..192670.422 rows=0 loops=1)
Workers Planned: 1
Params Evaluated: $0
Workers Launched: 1
-> Nested Loop Left Join (cost=0.00..91025107.47 rows=1
width=27) (actual time=192588.143..192588.144 rows=0 loops=2)
Join Filter: ($0 IS NOT NULL)
Filter: ((sample_0.l_orderkey IS NULL) AND
(sample_0.l_shipmode IS NULL))
Rows Removed by Filter: 1810515312
-> Parallel Seq Scan on lineitem ref_0 (cost=0.00..1721.97
rows=35397 width=11) (actual time=0.007..3.797 rows=30088 loops=2)
-> Seq Scan on lineitem sample_0 (cost=0.00..1969.75
rows=60175 width=27) (actual time=0.000..2.637 rows=60175 loops=60175)
Planning Time: 0.174 ms
Execution Time: 192670.458 ms
(19 rows)
I debug the code and find consider_parallel_nestloop() doesn't consider
materialized form of the cheapest inner path.
When enable_material = true, we can see Material path won in first plan,
but Parallel Seq Scan node doesn't add as outer path, which because
in try_partial_nestloop_path() , the cost of nestloop wat computed using
seq scan path not material path.
[1] include test table schema and data, you can repeat above problem.
I try fix this problem in attached patch, and I found pg12.12 also had this
issue. Please review my patch, thanks!
[1] https://github.com/tenderwg/tpch_test
From 7b652382928cc80ab6fbe782feada9f0eab4c5a8 Mon Sep 17 00:00:00 2001
From: "tender.wang" <[email protected]>
Date: Tue, 5 Sep 2023 14:33:24 +0800
Subject: [PATCH] Parallel seq scan should consider materila inner path in
nestloop case.
---
src/backend/optimizer/path/joinpath.c | 16 +++++++++++
src/test/regress/expected/select_parallel.out | 27 +++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 10 +++++++
3 files changed, 53 insertions(+)
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index 821d282497..5a10fb7f4b 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -2004,10 +2004,22 @@ consider_parallel_nestloop(PlannerInfo *root,
{
JoinType save_jointype = jointype;
ListCell *lc1;
+ Path *matpath = NULL;
+ Path *inner_cheapest_total = innerrel->cheapest_total_path;
if (jointype == JOIN_UNIQUE_INNER)
jointype = JOIN_INNER;
+ /*
+ * Consider materializing the cheapest inner path, unless
+ * enable_material is off or the path in question materializes its
+ * output anyway.
+ */
+ if (enable_material && inner_cheapest_total != NULL &&
+ !ExecMaterializesOutput(inner_cheapest_total->pathtype))
+ matpath = (Path *)
+ create_material_path(innerrel, inner_cheapest_total);
+
foreach(lc1, outerrel->partial_pathlist)
{
Path *outerpath = (Path *) lfirst(lc1);
@@ -2064,6 +2076,10 @@ consider_parallel_nestloop(PlannerInfo *root,
try_partial_nestloop_path(root, joinrel, outerpath, mpath,
pathkeys, jointype, extra);
}
+ /* Also consider materialized form of the cheapest inner path */
+ if (matpath != NULL && matpath->parallel_safe)
+ try_partial_nestloop_path(root, joinrel, outerpath, matpath,
+ pathkeys, jointype, extra);
}
}
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d88353d496..452a3aed07 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -844,6 +844,33 @@ select * from
(12 rows)
reset enable_material;
+-- test materialized form of the cheapest inner path
+set min_parallel_table_scan_size = '512kB';
+explain(costs off) select suq.two, suq.four
+from
+ (select t1.two, t2.four, t1.ten,t2.twenty
+ from tenk1 t1 left join tenk2 t2
+ on ((select q1 from int8_tbl order by q1 limit 1)is not null) where t2.ten is null) as suq;
+ QUERY PLAN
+-------------------------------------------------
+ Nested Loop Left Join
+ Join Filter: ($0 IS NOT NULL)
+ Filter: (t2.ten IS NULL)
+ InitPlan 1 (returns $0)
+ -> Limit
+ -> Sort
+ Sort Key: int8_tbl.q1
+ -> Seq Scan on int8_tbl
+ -> Gather
+ Workers Planned: 4
+ -> Parallel Seq Scan on tenk1 t1
+ -> Materialize
+ -> Gather
+ Workers Planned: 2
+ -> Parallel Seq Scan on tenk2 t2
+(15 rows)
+
+set min_parallel_table_scan_size = 0;
reset enable_hashagg;
-- check parallelized int8 aggregate (bug #14897)
explain (costs off)
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 80c914dc02..ded44a0c09 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -312,6 +312,16 @@ select * from
reset enable_material;
+-- test materialized form of the cheapest inner path
+set min_parallel_table_scan_size = '512kB';
+
+explain(costs off) select suq.two, suq.four
+from
+ (select t1.two, t2.four, t1.ten,t2.twenty
+ from tenk1 t1 left join tenk2 t2
+ on ((select q1 from int8_tbl order by q1 limit 1)is not null) where t2.ten is null) as suq;
+
+set min_parallel_table_scan_size = 0;
reset enable_hashagg;
-- check parallelized int8 aggregate (bug #14897)
--
2.25.1
Attachments:
[text/plain] 0001-Parallel-seq-scan-should-consider-materila-inner-pat.patch (3.8K, ../../CAHewXNkPmtEXNfVQMou_7NqQmFABca9f4etjBtdbbm0ZKDmWvw@mail.gmail.com/3-0001-Parallel-seq-scan-should-consider-materila-inner-pat.patch)
download | inline diff:
From 7b652382928cc80ab6fbe782feada9f0eab4c5a8 Mon Sep 17 00:00:00 2001
From: "tender.wang" <[email protected]>
Date: Tue, 5 Sep 2023 14:33:24 +0800
Subject: [PATCH] Parallel seq scan should consider materila inner path in
nestloop case.
---
src/backend/optimizer/path/joinpath.c | 16 +++++++++++
src/test/regress/expected/select_parallel.out | 27 +++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 10 +++++++
3 files changed, 53 insertions(+)
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index 821d282497..5a10fb7f4b 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -2004,10 +2004,22 @@ consider_parallel_nestloop(PlannerInfo *root,
{
JoinType save_jointype = jointype;
ListCell *lc1;
+ Path *matpath = NULL;
+ Path *inner_cheapest_total = innerrel->cheapest_total_path;
if (jointype == JOIN_UNIQUE_INNER)
jointype = JOIN_INNER;
+ /*
+ * Consider materializing the cheapest inner path, unless
+ * enable_material is off or the path in question materializes its
+ * output anyway.
+ */
+ if (enable_material && inner_cheapest_total != NULL &&
+ !ExecMaterializesOutput(inner_cheapest_total->pathtype))
+ matpath = (Path *)
+ create_material_path(innerrel, inner_cheapest_total);
+
foreach(lc1, outerrel->partial_pathlist)
{
Path *outerpath = (Path *) lfirst(lc1);
@@ -2064,6 +2076,10 @@ consider_parallel_nestloop(PlannerInfo *root,
try_partial_nestloop_path(root, joinrel, outerpath, mpath,
pathkeys, jointype, extra);
}
+ /* Also consider materialized form of the cheapest inner path */
+ if (matpath != NULL && matpath->parallel_safe)
+ try_partial_nestloop_path(root, joinrel, outerpath, matpath,
+ pathkeys, jointype, extra);
}
}
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d88353d496..452a3aed07 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -844,6 +844,33 @@ select * from
(12 rows)
reset enable_material;
+-- test materialized form of the cheapest inner path
+set min_parallel_table_scan_size = '512kB';
+explain(costs off) select suq.two, suq.four
+from
+ (select t1.two, t2.four, t1.ten,t2.twenty
+ from tenk1 t1 left join tenk2 t2
+ on ((select q1 from int8_tbl order by q1 limit 1)is not null) where t2.ten is null) as suq;
+ QUERY PLAN
+-------------------------------------------------
+ Nested Loop Left Join
+ Join Filter: ($0 IS NOT NULL)
+ Filter: (t2.ten IS NULL)
+ InitPlan 1 (returns $0)
+ -> Limit
+ -> Sort
+ Sort Key: int8_tbl.q1
+ -> Seq Scan on int8_tbl
+ -> Gather
+ Workers Planned: 4
+ -> Parallel Seq Scan on tenk1 t1
+ -> Materialize
+ -> Gather
+ Workers Planned: 2
+ -> Parallel Seq Scan on tenk2 t2
+(15 rows)
+
+set min_parallel_table_scan_size = 0;
reset enable_hashagg;
-- check parallelized int8 aggregate (bug #14897)
explain (costs off)
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 80c914dc02..ded44a0c09 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -312,6 +312,16 @@ select * from
reset enable_material;
+-- test materialized form of the cheapest inner path
+set min_parallel_table_scan_size = '512kB';
+
+explain(costs off) select suq.two, suq.four
+from
+ (select t1.two, t2.four, t1.ten,t2.twenty
+ from tenk1 t1 left join tenk2 t2
+ on ((select q1 from int8_tbl order by q1 limit 1)is not null) where t2.ten is null) as suq;
+
+set min_parallel_table_scan_size = 0;
reset enable_hashagg;
-- check parallelized int8 aggregate (bug #14897)
--
2.25.1
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Should consider materializing the cheapest inner path in consider_parallel_nestloop()
@ 2023-09-05 10:50 Richard Guo <[email protected]>
parent: tender wang <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Richard Guo @ 2023-09-05 10:50 UTC (permalink / raw)
To: tender wang <[email protected]>; +Cc: [email protected]
On Tue, Sep 5, 2023 at 4:52 PM tender wang <[email protected]> wrote:
> I recently run benchmark[1] on master, but I found performance problem
> as below:
> ...
>
> I debug the code and find consider_parallel_nestloop() doesn't consider
> materialized form of the cheapest inner path.
>
Yeah, this seems an omission in commit 45be99f8. I reviewed the patch
and here are some comments.
* I think we should not consider materializing the cheapest inner path
if we're doing JOIN_UNIQUE_INNER, because in this case we have to
unique-ify the inner path.
* I think we can check if it'd be parallel safe before creating the
material path, thus avoid the creation in unsafe cases.
* I don't think the test case you added works for the code changes.
Maybe a plan likes below is better:
explain (costs off)
select * from tenk1, tenk2 where tenk1.two = tenk2.two;
QUERY PLAN
----------------------------------------------
Gather
Workers Planned: 4
-> Nested Loop
Join Filter: (tenk1.two = tenk2.two)
-> Parallel Seq Scan on tenk1
-> Materialize
-> Seq Scan on tenk2
(7 rows)
Thanks
Richard
^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2023-09-05 10:50 UTC | newest]
Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-12-21 06:11 [PATCH 08/20] union{} with a CompressionAlgorithm alg Justin Pryzby <[email protected]>
2023-09-05 08:52 Should consider materializing the cheapest inner path in consider_parallel_nestloop() tender wang <[email protected]>
2023-09-05 10:50 ` Re: Should consider materializing the cheapest inner path in consider_parallel_nestloop() Richard Guo <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox