public inbox for [email protected]  
help / color / mirror / Atom feed
[PATCH 7/8] add wal_compression_method=zstd
4+ messages / 3 participants
[nested] [flat]

* [PATCH 7/8] add wal_compression_method=zstd
@ 2021-03-12 20:43  Justin Pryzby <[email protected]>
  0 siblings, 0 replies; 4+ messages in thread

From: Justin Pryzby @ 2021-03-12 20:43 UTC (permalink / raw)

---
 configure                               | 163 ++++++++++++++++++++++++
 configure.ac                            |  26 ++++
 src/backend/access/transam/xlog.c       |   3 +
 src/backend/access/transam/xloginsert.c |  20 +++
 src/backend/access/transam/xlogreader.c |  15 +++
 src/include/access/xlog_internal.h      |   1 +
 src/include/pg_config.h.in              |   3 +
 src/tools/msvc/Solution.pm              |   1 +
 8 files changed, 232 insertions(+)

diff --git a/configure b/configure
index 3afa1e87e3..c7c4c951a7 100755
--- a/configure
+++ b/configure
@@ -699,6 +699,9 @@ with_gnu_ld
 LD
 LDFLAGS_SL
 LDFLAGS_EX
+ZSTD_LIBS
+ZSTD_CFLAGS
+with_zstd
 LZ4_LIBS
 LZ4_CFLAGS
 with_lz4
@@ -868,6 +871,7 @@ with_libxslt
 with_system_tzdata
 with_zlib
 with_lz4
+with_zstd
 with_gnu_ld
 with_ssl
 with_openssl
@@ -897,6 +901,8 @@ XML2_CFLAGS
 XML2_LIBS
 LZ4_CFLAGS
 LZ4_LIBS
+ZSTD_CFLAGS
+ZSTD_LIBS
 LDFLAGS_EX
 LDFLAGS_SL
 PERL
@@ -1576,6 +1582,7 @@ Optional Packages:
                           use system time zone data in DIR
   --without-zlib          do not use Zlib
   --without-lz4           build without LZ4 support
+  --with-zstd             build with zstd support
   --with-gnu-ld           assume the C compiler uses GNU ld [default=no]
   --with-ssl=LIB          use LIB for SSL/TLS support (openssl)
   --with-openssl          obsolete spelling of --with-ssl=openssl
@@ -1605,6 +1612,8 @@ Some influential environment variables:
   XML2_LIBS   linker flags for XML2, overriding pkg-config
   LZ4_CFLAGS  C compiler flags for LZ4, overriding pkg-config
   LZ4_LIBS    linker flags for LZ4, overriding pkg-config
+  ZSTD_CFLAGS C compiler flags for ZSTD, overriding pkg-config
+  ZSTD_LIBS   linker flags for ZSTD, overriding pkg-config
   LDFLAGS_EX  extra linker flags for linking executables only
   LDFLAGS_SL  extra linker flags for linking shared libraries only
   PERL        Perl program
@@ -8705,6 +8714,137 @@ fi
   CFLAGS="$LZ4_CFLAGS $CFLAGS"
 fi
 
+#
+# ZSTD
+#
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking whether to build with zstd support" >&5
+$as_echo_n "checking whether to build with zstd support... " >&6; }
+
+
+
+# Check whether --with-zstd was given.
+if test "${with_zstd+set}" = set; then :
+  withval=$with_zstd;
+  case $withval in
+    yes)
+
+$as_echo "#define USE_ZSTD 1" >>confdefs.h
+
+      ;;
+    no)
+      :
+      ;;
+    *)
+      as_fn_error $? "no argument expected for --with-zstd option" "$LINENO" 5
+      ;;
+  esac
+
+else
+  with_zstd=no
+
+fi
+
+
+{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $with_zstd" >&5
+$as_echo "$with_zstd" >&6; }
+
+
+if test "$with_zstd" = yes; then
+
+pkg_failed=no
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking for libzstd" >&5
+$as_echo_n "checking for libzstd... " >&6; }
+
+if test -n "$ZSTD_CFLAGS"; then
+    pkg_cv_ZSTD_CFLAGS="$ZSTD_CFLAGS"
+ elif test -n "$PKG_CONFIG"; then
+    if test -n "$PKG_CONFIG" && \
+    { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"libzstd\""; } >&5
+  ($PKG_CONFIG --exists --print-errors "libzstd") 2>&5
+  ac_status=$?
+  $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5
+  test $ac_status = 0; }; then
+  pkg_cv_ZSTD_CFLAGS=`$PKG_CONFIG --cflags "libzstd" 2>/dev/null`
+		      test "x$?" != "x0" && pkg_failed=yes
+else
+  pkg_failed=yes
+fi
+ else
+    pkg_failed=untried
+fi
+if test -n "$ZSTD_LIBS"; then
+    pkg_cv_ZSTD_LIBS="$ZSTD_LIBS"
+ elif test -n "$PKG_CONFIG"; then
+    if test -n "$PKG_CONFIG" && \
+    { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"libzstd\""; } >&5
+  ($PKG_CONFIG --exists --print-errors "libzstd") 2>&5
+  ac_status=$?
+  $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5
+  test $ac_status = 0; }; then
+  pkg_cv_ZSTD_LIBS=`$PKG_CONFIG --libs "libzstd" 2>/dev/null`
+		      test "x$?" != "x0" && pkg_failed=yes
+else
+  pkg_failed=yes
+fi
+ else
+    pkg_failed=untried
+fi
+
+
+
+if test $pkg_failed = yes; then
+        { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+
+if $PKG_CONFIG --atleast-pkgconfig-version 0.20; then
+        _pkg_short_errors_supported=yes
+else
+        _pkg_short_errors_supported=no
+fi
+        if test $_pkg_short_errors_supported = yes; then
+	        ZSTD_PKG_ERRORS=`$PKG_CONFIG --short-errors --print-errors --cflags --libs "libzstd" 2>&1`
+        else
+	        ZSTD_PKG_ERRORS=`$PKG_CONFIG --print-errors --cflags --libs "libzstd" 2>&1`
+        fi
+	# Put the nasty error message in config.log where it belongs
+	echo "$ZSTD_PKG_ERRORS" >&5
+
+	as_fn_error $? "Package requirements (libzstd) were not met:
+
+$ZSTD_PKG_ERRORS
+
+Consider adjusting the PKG_CONFIG_PATH environment variable if you
+installed software in a non-standard prefix.
+
+Alternatively, you may set the environment variables ZSTD_CFLAGS
+and ZSTD_LIBS to avoid the need to call pkg-config.
+See the pkg-config man page for more details." "$LINENO" 5
+elif test $pkg_failed = untried; then
+        { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+	{ { $as_echo "$as_me:${as_lineno-$LINENO}: error: in \`$ac_pwd':" >&5
+$as_echo "$as_me: error: in \`$ac_pwd':" >&2;}
+as_fn_error $? "The pkg-config script could not be found or is too old.  Make sure it
+is in your PATH or set the PKG_CONFIG environment variable to the full
+path to pkg-config.
+
+Alternatively, you may set the environment variables ZSTD_CFLAGS
+and ZSTD_LIBS to avoid the need to call pkg-config.
+See the pkg-config man page for more details.
+
+To get pkg-config, see <http://pkg-config.freedesktop.org/;.
+See \`config.log' for more details" "$LINENO" 5; }
+else
+	ZSTD_CFLAGS=$pkg_cv_ZSTD_CFLAGS
+	ZSTD_LIBS=$pkg_cv_ZSTD_LIBS
+        { $as_echo "$as_me:${as_lineno-$LINENO}: result: yes" >&5
+$as_echo "yes" >&6; }
+
+fi
+  LIBS="$ZSTD_LIBS $LIBS"
+  CFLAGS="$ZSTD_CFLAGS $CFLAGS"
+fi
+
 #
 # Assignments
 #
@@ -13556,6 +13696,29 @@ done
   CPPFLAGS=$ac_save_CPPFLAGS
 fi
 
+if test "$with_zstd" = yes; then
+  ac_save_CPPFLAGS=$CPPFLAGS
+  CPPFLAGS="$ZSTD_CFLAGS $CPPFLAGS"
+
+  # Verify we have zstd's header files
+  for ac_header in zstd.h
+do :
+  ac_fn_c_check_header_mongrel "$LINENO" "zstd.h" "ac_cv_header_zstd_h" "$ac_includes_default"
+if test "x$ac_cv_header_zstd_h" = xyes; then :
+  cat >>confdefs.h <<_ACEOF
+#define HAVE_ZSTD_H 1
+_ACEOF
+
+else
+  as_fn_error $? "zstd.h header file is required for zstd" "$LINENO" 5
+fi
+
+done
+
+
+  CPPFLAGS=$ac_save_CPPFLAGS
+fi
+
 if test "$with_gssapi" = yes ; then
   for ac_header in gssapi/gssapi.h
 do :
diff --git a/configure.ac b/configure.ac
index d62ea5f742..abac4b28f6 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1001,6 +1001,21 @@ if test "$with_lz4" = yes; then
   CFLAGS="$LZ4_CFLAGS $CFLAGS"
 fi
 
+#
+# ZSTD
+#
+AC_MSG_CHECKING([whether to build with zstd support])
+PGAC_ARG_BOOL(with, zstd, no, [build with zstd support],
+              [AC_DEFINE([USE_ZSTD], 1, [Define to 1 to build with zstd support. (--with-zstd)])])
+AC_MSG_RESULT([$with_zstd])
+AC_SUBST(with_zstd)
+
+if test "$with_zstd" = yes; then
+  PKG_CHECK_MODULES(ZSTD, libzstd)
+  LIBS="$ZSTD_LIBS $LIBS"
+  CFLAGS="$ZSTD_CFLAGS $CFLAGS"
+fi
+
 #
 # Assignments
 #
@@ -1433,6 +1448,17 @@ if test "$with_lz4" = yes; then
   CPPFLAGS=$ac_save_CPPFLAGS
 fi
 
+if test "$with_zstd" = yes; then
+  ac_save_CPPFLAGS=$CPPFLAGS
+  CPPFLAGS="$ZSTD_CFLAGS $CPPFLAGS"
+
+  # Verify we have zstd's header files
+  AC_CHECK_HEADERS(zstd.h, [],
+       [AC_MSG_ERROR([zstd.h header file is required for zstd])])
+
+  CPPFLAGS=$ac_save_CPPFLAGS
+fi
+
 if test "$with_gssapi" = yes ; then
   AC_CHECK_HEADERS(gssapi/gssapi.h, [],
 	[AC_CHECK_HEADERS(gssapi.h, [], [AC_MSG_ERROR([gssapi.h header file is required for GSSAPI])])])
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f33355e44a..f2749ebf45 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -189,6 +189,9 @@ const struct config_enum_entry wal_compression_options[] = {
 #endif
 #ifdef  USE_LZ4
 	{"lz4", WAL_COMPRESSION_LZ4, false},
+#endif
+#ifdef  USE_ZSTD
+	{"zstd", WAL_COMPRESSION_ZSTD, false},
 #endif
 	{NULL, 0, false}
 };
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 38ac5f9e94..852e5ef7af 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -41,6 +41,10 @@
 #include "lz4.h"
 #endif
 
+#ifdef USE_ZSTD
+#include "zstd.h"
+#endif
+
 /* Buffer size required to store a compressed version of backup block image */
 #define PGLZ_MAX_BLCKSZ PGLZ_MAX_OUTPUT(BLCKSZ)
 
@@ -896,6 +900,22 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
 		break;
 #endif
 
+#ifdef USE_ZSTD
+	case WAL_COMPRESSION_ZSTD:
+		len = ZSTD_compress(dest, PGLZ_MAX_BLCKSZ,
+				source, orig_len, ZSTD_CLEVEL_DEFAULT);
+		if (ZSTD_isError(len))
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("failed compressing zstd: %s",
+						 ZSTD_getErrorName(len))));
+			len = -1;
+		}
+
+		break;
+#endif
+
 	default:
 		/*
 		 * It should be impossible to get here for unsupported algorithms,
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 920c47e4d7..6ce2fbfb13 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -41,6 +41,10 @@
 #include "lz4.h"
 #endif
 
+#ifdef USE_ZSTD
+#include "zstd.h"
+#endif
+
 static void report_invalid_record(XLogReaderState *state, const char *fmt,...)
 			pg_attribute_printf(2, 3);
 static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength);
@@ -1564,6 +1568,8 @@ wal_compression_name(WalCompression compression)
 			return "zlib";
 		case WAL_COMPRESSION_LZ4:
 			return "lz4";
+		case WAL_COMPRESSION_ZSTD:
+			return "zstd";
 		default:
 			return "???";
 	}
@@ -1619,6 +1625,15 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
 			break;
 #endif
 
+#ifdef USE_ZSTD
+		case WAL_COMPRESSION_ZSTD:
+			decomp_result = ZSTD_decompress(tmp.data, BLCKSZ, ptr, bkpb->bimg_len);
+			// XXX: ZSTD_getErrorName
+			if (ZSTD_isError(decomp_result))
+				decomp_result = -1;
+			break;
+#endif
+
 		default:
 			report_invalid_record(record, "image at %X/%X is compressed with unsupported codec, block %d (%d/%s)",
 								  (uint32) (record->ReadRecPtr >> 32),
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index 0993de784e..30f663f448 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -329,6 +329,7 @@ typedef enum WalCompression
 	WAL_COMPRESSION_PGLZ = 0,
 	WAL_COMPRESSION_ZLIB = 1,
 	WAL_COMPRESSION_LZ4 = 2,
+	WAL_COMPRESSION_ZSTD = 3,
 } WalCompression;
 
 extern const char *wal_compression_name(WalCompression compression);
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index 0a6422da4f..ad26393352 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -902,6 +902,9 @@
 /* Define to 1 to build with LZ4 support (--with-lz4) */
 #undef USE_LZ4
 
+/* Define to 1 if you have the `zstd' library (-lzstd). */
+#undef USE_ZSTD
+
 /* Define to select named POSIX semaphores. */
 #undef USE_NAMED_POSIX_SEMAPHORES
 
diff --git a/src/tools/msvc/Solution.pm b/src/tools/msvc/Solution.pm
index 14605371bb..7dfa0f31f6 100644
--- a/src/tools/msvc/Solution.pm
+++ b/src/tools/msvc/Solution.pm
@@ -307,6 +307,7 @@ sub GenerateFiles
 		HAVE_LIBXML2                                => undef,
 		HAVE_LIBXSLT                                => undef,
 		HAVE_LIBZ                   => $self->{options}->{zlib} ? 1 : undef,
+		HAVE_LIBZSTD                => $self->{options}->{zstd} ? 1 : undef,
 		HAVE_LINK                   => undef,
 		HAVE_LOCALE_T               => 1,
 		HAVE_LONG_INT_64            => undef,
-- 
2.17.0


--f0KYrhQ4vYSV2aJu--





^ permalink  raw  reply  [nested|flat] 4+ messages in thread

* Re: Use read streams in CREATE DATABASE command when the strategy is wal_log
@ 2024-07-16 11:11  Nazir Bilal Yavuz <[email protected]>
  0 siblings, 1 reply; 4+ messages in thread

From: Nazir Bilal Yavuz @ 2024-07-16 11:11 UTC (permalink / raw)
  To: Noah Misch <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>

Hi,

Thank you for the review!

On Fri, 12 Jul 2024 at 02:52, Noah Misch <[email protected]> wrote:
>
> On Tue, Apr 16, 2024 at 02:12:19PM +0300, Nazir Bilal Yavuz wrote:
> > I am working on using read streams in the CREATE DATABASE command when the
> > strategy is wal_log. RelationCopyStorageUsingBuffer() function is used in
> > this context. This function reads source buffers then copies them to the
>
> Please rebase.  I applied this to 40126ac for review purposes.

Rebased.

> > Subject: [PATCH v1 1/3] Refactor PinBufferForBlock() to remove if checks about
> >  persistence
> >
> > There are if checks in PinBufferForBlock() function to set persistence
> > of the relation and this function is called for the each block in the
> > relation. Instead of that, set persistence of the relation before
> > PinBufferForBlock() function.
>
> I tried with the following additional patch to see if PinBufferForBlock() ever
> gets invalid smgr_relpersistence:
>
> ====
> --- a/src/backend/storage/buffer/bufmgr.c
> +++ b/src/backend/storage/buffer/bufmgr.c
> @@ -1098,6 +1098,11 @@ PinBufferForBlock(Relation rel,
>
>         Assert(blockNum != P_NEW);
>
> +       if (!(smgr_persistence == RELPERSISTENCE_TEMP ||
> +                 smgr_persistence == RELPERSISTENCE_PERMANENT ||
> +                 smgr_persistence == RELPERSISTENCE_UNLOGGED))
> +               elog(WARNING, "unexpected relpersistence %d", smgr_persistence);
> +
>         if (smgr_persistence == RELPERSISTENCE_TEMP)
>         {
>                 io_context = IOCONTEXT_NORMAL;
> ====
>
> That still gets relpersistence==0 in various src/test/regress cases.  I think
> the intent was to prevent that.  If not, please add a comment about when
> relpersistence==0 is still allowed.

I fixed it, it is caused by (mode == RBM_ZERO_AND_CLEANUP_LOCK | mode
== RBM_ZERO_AND_LOCK) case in the ReadBuffer_common(). The persistence
was not updated for this path before. I also added an assert check for
this problem to PinBufferForBlock().

> > --- a/src/backend/storage/aio/read_stream.c
> > +++ b/src/backend/storage/aio/read_stream.c
> > @@ -549,7 +549,7 @@ read_stream_begin_relation(int flags,
> >       {
> >               stream->ios[i].op.rel = rel;
> >               stream->ios[i].op.smgr = RelationGetSmgr(rel);
> > -             stream->ios[i].op.smgr_persistence = 0;
> > +             stream->ios[i].op.smgr_persistence = rel->rd_rel->relpersistence;
>
> Does the following comment in ReadBuffersOperation need an update?
>
>         /*
>          * The following members should be set by the caller.  If only smgr is
>          * provided without rel, then smgr_persistence can be set to override the
>          * default assumption of RELPERSISTENCE_PERMANENT.
>          */
>

I believe it does not need to be updated but I renamed
'ReadBuffersOperation.smgr_persistence' as
'ReadBuffersOperation.persistence'. So, this comment is updated as
well. I think that rename suits better because persistence does not
need to come from smgr, it could come from relation, too. Do you think
it is a good idea? If it is, does it need a separate commit?

> > --- a/src/backend/storage/buffer/bufmgr.c
> > +++ b/src/backend/storage/buffer/bufmgr.c
>
> > +/*
> > + * Helper struct for read stream object used in
> > + * RelationCopyStorageUsingBuffer() function.
> > + */
> > +struct copy_storage_using_buffer_read_stream_private
> > +{
> > +     BlockNumber blocknum;
> > +     int64           last_block;
> > +};
>
> Why is last_block an int64, not a BlockNumber?

You are right, the type of last_block should be BlockNumber; done. I
copied it from pg_prewarm_read_stream_private struct and I guess the
same should be applied to it as well but it is not the topic of this
thread, so I did not update it yet.

> > @@ -4667,19 +4698,31 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
>
> >       /* Iterate over each block of the source relation file. */
> >       for (blkno = 0; blkno < nblocks; blkno++)
> >       {
> >               CHECK_FOR_INTERRUPTS();
> >
> >               /* Read block from source relation. */
> > -             srcBuf = ReadBufferWithoutRelcache(srclocator, forkNum, blkno,
> > -                                                                                RBM_NORMAL, bstrategy_src,
> > -                                                                                permanent);
> > +             srcBuf = read_stream_next_buffer(src_stream, NULL);
> >               LockBuffer(srcBuf, BUFFER_LOCK_SHARE);
>
> I think this should check for read_stream_next_buffer() returning
> InvalidBuffer.  pg_prewarm doesn't, but the other callers do, and I think the
> other callers are a better model.  LockBuffer() doesn't check the
> InvalidBuffer case, so let's avoid the style of using a
> read_stream_next_buffer() return value without checking.

There is an assert in the LockBuffer which checks for the
InvalidBuffer. If that is not enough, we may add an if check for
InvalidBuffer but what should we do in this case? It should not
happen, so erroring out may be a good idea.

Updated patches are attached (without InvalidBuffer check for now).

--
Regards,
Nazir Bilal Yavuz
Microsoft


Attachments:

  [text/x-patch] v2-0001-Refactor-PinBufferForBlock-to-remove-if-checks-ab.patch (4.8K, ../../CAN55FZ0pHWZrLPwKGggnigJCj9Cp7fCXY8-e5yGNLzzp5kPvgg@mail.gmail.com/2-v2-0001-Refactor-PinBufferForBlock-to-remove-if-checks-ab.patch)
  download | inline diff:
From 977cea9451e602d8d3d5e4f0cf3cd7dfea11879e Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Sun, 7 Apr 2024 22:33:36 +0300
Subject: [PATCH v2 1/3] Refactor PinBufferForBlock() to remove if checks about
 persistence

There are if checks in PinBufferForBlock() function to set persistence
of the relation and this function is called for the each block in the
relation. Instead of that, set persistence of the relation before
PinBufferForBlock() function.
---
 src/include/storage/bufmgr.h          |  4 ++--
 src/backend/storage/aio/read_stream.c |  2 +-
 src/backend/storage/buffer/bufmgr.c   | 32 +++++++++++++--------------
 3 files changed, 18 insertions(+), 20 deletions(-)

diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index a1e71013d32..ba9a1a289a9 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -116,12 +116,12 @@ struct ReadBuffersOperation
 {
 	/*
 	 * The following members should be set by the caller.  If only smgr is
-	 * provided without rel, then smgr_persistence can be set to override the
+	 * provided without rel, then persistence can be set to override the
 	 * default assumption of RELPERSISTENCE_PERMANENT.
 	 */
 	Relation	rel;
 	struct SMgrRelationData *smgr;
-	char		smgr_persistence;
+	char		persistence;
 	ForkNumber	forknum;
 	BufferAccessStrategy strategy;
 
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 74b9bae6313..58221649f27 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -551,7 +551,7 @@ read_stream_begin_relation(int flags,
 	{
 		stream->ios[i].op.rel = rel;
 		stream->ios[i].op.smgr = RelationGetSmgr(rel);
-		stream->ios[i].op.smgr_persistence = 0;
+		stream->ios[i].op.persistence = rel->rd_rel->relpersistence;
 		stream->ios[i].op.forknum = forknum;
 		stream->ios[i].op.strategy = strategy;
 	}
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 61816730955..7cbcdc63a6f 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -1104,7 +1104,7 @@ ZeroAndLockBuffer(Buffer buffer, ReadBufferMode mode, bool already_valid)
 static pg_attribute_always_inline Buffer
 PinBufferForBlock(Relation rel,
 				  SMgrRelation smgr,
-				  char smgr_persistence,
+				  char persistence,
 				  ForkNumber forkNum,
 				  BlockNumber blockNum,
 				  BufferAccessStrategy strategy,
@@ -1113,22 +1113,12 @@ PinBufferForBlock(Relation rel,
 	BufferDesc *bufHdr;
 	IOContext	io_context;
 	IOObject	io_object;
-	char		persistence;
 
 	Assert(blockNum != P_NEW);
 
-	/*
-	 * If there is no Relation it usually implies recovery and thus permanent,
-	 * but we take an argument because CreateAndCopyRelationData can reach us
-	 * with only an SMgrRelation for an unlogged relation that we don't want
-	 * to flag with BM_PERMANENT.
-	 */
-	if (rel)
-		persistence = rel->rd_rel->relpersistence;
-	else if (smgr_persistence == 0)
-		persistence = RELPERSISTENCE_PERMANENT;
-	else
-		persistence = smgr_persistence;
+	Assert((persistence == RELPERSISTENCE_TEMP ||
+			persistence == RELPERSISTENCE_PERMANENT ||
+			persistence == RELPERSISTENCE_UNLOGGED));
 
 	if (persistence == RELPERSISTENCE_TEMP)
 	{
@@ -1203,6 +1193,7 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 	ReadBuffersOperation operation;
 	Buffer		buffer;
 	int			flags;
+	char		persistence;
 
 	/*
 	 * Backward compatibility path, most code should use ExtendBufferedRel()
@@ -1224,12 +1215,19 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 		return ExtendBufferedRel(BMR_REL(rel), forkNum, strategy, flags);
 	}
 
+	if (rel)
+		persistence = rel->rd_rel->relpersistence;
+	else if (smgr_persistence == 0)
+		persistence = RELPERSISTENCE_PERMANENT;
+	else
+		persistence = smgr_persistence;
+
 	if (unlikely(mode == RBM_ZERO_AND_CLEANUP_LOCK ||
 				 mode == RBM_ZERO_AND_LOCK))
 	{
 		bool		found;
 
-		buffer = PinBufferForBlock(rel, smgr, smgr_persistence,
+		buffer = PinBufferForBlock(rel, smgr, persistence,
 								   forkNum, blockNum, strategy, &found);
 		ZeroAndLockBuffer(buffer, mode, found);
 		return buffer;
@@ -1241,7 +1239,7 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 		flags = 0;
 	operation.smgr = smgr;
 	operation.rel = rel;
-	operation.smgr_persistence = smgr_persistence;
+	operation.persistence = persistence;
 	operation.forknum = forkNum;
 	operation.strategy = strategy;
 	if (StartReadBuffer(&operation,
@@ -1272,7 +1270,7 @@ StartReadBuffersImpl(ReadBuffersOperation *operation,
 
 		buffers[i] = PinBufferForBlock(operation->rel,
 									   operation->smgr,
-									   operation->smgr_persistence,
+									   operation->persistence,
 									   operation->forknum,
 									   blockNum + i,
 									   operation->strategy,
-- 
2.45.2



  [text/x-patch] v2-0002-Add-a-way-to-create-read-stream-object-by-using-S.patch (5.7K, ../../CAN55FZ0pHWZrLPwKGggnigJCj9Cp7fCXY8-e5yGNLzzp5kPvgg@mail.gmail.com/3-v2-0002-Add-a-way-to-create-read-stream-object-by-using-S.patch)
  download | inline diff:
From 53c2bd12d71e521d975002fb03ef53b74389ccb7 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Mon, 8 Apr 2024 00:14:52 +0300
Subject: [PATCH v2 2/3] Add a way to create read stream object by using
 SMgrRelation

Currently read stream object can be created only by using Relation,
there is no way to create it by using SMgrRelation.

To achieve that, read_stream_begin_impl() function is created and
contents of the read_stream_begin_relation() is moved into this
function.

Both read_stream_begin_relation() and read_stream_begin_smgr_relation()
calls read_stream_begin_impl() by Relation and SMgrRelation
respectively.
---
 src/include/storage/read_stream.h     | 10 ++++
 src/backend/storage/aio/read_stream.c | 83 ++++++++++++++++++++++-----
 2 files changed, 79 insertions(+), 14 deletions(-)

diff --git a/src/include/storage/read_stream.h b/src/include/storage/read_stream.h
index f676d2cc20a..4e599904f26 100644
--- a/src/include/storage/read_stream.h
+++ b/src/include/storage/read_stream.h
@@ -15,6 +15,7 @@
 #define READ_STREAM_H
 
 #include "storage/bufmgr.h"
+#include "storage/smgr.h"
 
 /* Default tuning, reasonable for many users. */
 #define READ_STREAM_DEFAULT 0x00
@@ -57,6 +58,15 @@ extern ReadStream *read_stream_begin_relation(int flags,
 											  void *callback_private_data,
 											  size_t per_buffer_data_size);
 extern Buffer read_stream_next_buffer(ReadStream *stream, void **per_buffer_data);
+extern ReadStream *read_stream_begin_smgr_relation(int flags,
+												   BufferAccessStrategy strategy,
+												   SMgrRelation smgr,
+												   char smgr_persistence,
+												   ForkNumber forknum,
+												   ReadStreamBlockNumberCB callback,
+												   void *callback_private_data,
+												   size_t per_buffer_data_size);
+extern Buffer read_stream_next_buffer(ReadStream *stream, void **per_buffer_private);
 extern void read_stream_reset(ReadStream *stream);
 extern void read_stream_end(ReadStream *stream);
 
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 58221649f27..cdec918d29d 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -406,14 +406,16 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
  * write extra data for each block into the space provided to it.  It will
  * also receive callback_private_data for its own purposes.
  */
-ReadStream *
-read_stream_begin_relation(int flags,
-						   BufferAccessStrategy strategy,
-						   Relation rel,
-						   ForkNumber forknum,
-						   ReadStreamBlockNumberCB callback,
-						   void *callback_private_data,
-						   size_t per_buffer_data_size)
+static ReadStream *
+read_stream_begin_impl(int flags,
+					   BufferAccessStrategy strategy,
+					   Relation rel,
+					   SMgrRelation smgr,
+					   char smgr_persistence,
+					   ForkNumber forknum,
+					   ReadStreamBlockNumberCB callback,
+					   void *callback_private_data,
+					   size_t per_buffer_data_size)
 {
 	ReadStream *stream;
 	size_t		size;
@@ -422,9 +424,6 @@ read_stream_begin_relation(int flags,
 	int			strategy_pin_limit;
 	uint32		max_pinned_buffers;
 	Oid			tablespace_id;
-	SMgrRelation smgr;
-
-	smgr = RelationGetSmgr(rel);
 
 	/*
 	 * Decide how many I/Os we will allow to run at the same time.  That
@@ -434,7 +433,7 @@ read_stream_begin_relation(int flags,
 	 */
 	tablespace_id = smgr->smgr_rlocator.locator.spcOid;
 	if (!OidIsValid(MyDatabaseId) ||
-		IsCatalogRelation(rel) ||
+		(rel && IsCatalogRelation(rel)) ||
 		IsCatalogRelationOid(smgr->smgr_rlocator.locator.relNumber))
 	{
 		/*
@@ -550,8 +549,8 @@ read_stream_begin_relation(int flags,
 	for (int i = 0; i < max_ios; ++i)
 	{
 		stream->ios[i].op.rel = rel;
-		stream->ios[i].op.smgr = RelationGetSmgr(rel);
-		stream->ios[i].op.persistence = rel->rd_rel->relpersistence;
+		stream->ios[i].op.smgr = smgr;
+		stream->ios[i].op.persistence = smgr_persistence;
 		stream->ios[i].op.forknum = forknum;
 		stream->ios[i].op.strategy = strategy;
 	}
@@ -559,6 +558,62 @@ read_stream_begin_relation(int flags,
 	return stream;
 }
 
+/*
+ * Create a new read stream for reading a relation.
+ * See read_stream_begin_impl() for the detailed explanation.
+ */
+ReadStream *
+read_stream_begin_relation(int flags,
+						   BufferAccessStrategy strategy,
+						   Relation rel,
+						   ForkNumber forknum,
+						   ReadStreamBlockNumberCB callback,
+						   void *callback_private_data,
+						   size_t per_buffer_data_size)
+{
+	return read_stream_begin_impl(flags,
+								  strategy,
+								  rel,
+								  RelationGetSmgr(rel),
+								  rel->rd_rel->relpersistence,
+								  forknum,
+								  callback,
+								  callback_private_data,
+								  per_buffer_data_size);
+}
+
+/*
+ * Create a new read stream for reading a SMgr relation.
+ * See read_stream_begin_impl() for the detailed explanation.
+ */
+ReadStream *
+read_stream_begin_smgr_relation(int flags,
+								BufferAccessStrategy strategy,
+								SMgrRelation smgr,
+								char smgr_persistence,
+								ForkNumber forknum,
+								ReadStreamBlockNumberCB callback,
+								void *callback_private_data,
+								size_t per_buffer_data_size)
+{
+	char		persistence;
+
+	if (smgr_persistence == 0)
+		persistence = RELPERSISTENCE_PERMANENT;
+	else
+		persistence = smgr_persistence;
+
+	return read_stream_begin_impl(flags,
+								  strategy,
+								  NULL,
+								  smgr,
+								  persistence,
+								  forknum,
+								  callback,
+								  callback_private_data,
+								  per_buffer_data_size);
+}
+
 /*
  * Pull one pinned buffer out of a stream.  Each call returns successive
  * blocks in the order specified by the callback.  If per_buffer_data_size was
-- 
2.45.2



  [text/x-patch] v2-0003-Use-read-streams-in-CREATE-DATABASE-when-strategy.patch (4.0K, ../../CAN55FZ0pHWZrLPwKGggnigJCj9Cp7fCXY8-e5yGNLzzp5kPvgg@mail.gmail.com/4-v2-0003-Use-read-streams-in-CREATE-DATABASE-when-strategy.patch)
  download | inline diff:
From 31a2d8a7eef71bcd56804535f4e8afa90c8e86d3 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Tue, 16 Jul 2024 13:22:15 +0300
Subject: [PATCH v2 3/3] Use read streams in CREATE DATABASE when strategy is
 wal_log

CREATE DABASE command uses RelationCopyStorageUsingBuffer() function to
copy buffers when the strategy is wal_log. This function reads source
buffers then copies them to the destination buffers. Read streams are
used only only while reading source buffers because the destination
buffer is read by 'RBM_ZERO_AND_LOCK' option, so it is not important.
---
 src/backend/storage/buffer/bufmgr.c | 53 ++++++++++++++++++++++++++---
 1 file changed, 49 insertions(+), 4 deletions(-)

diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 7cbcdc63a6f..2acc109c4db 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -54,6 +54,7 @@
 #include "storage/ipc.h"
 #include "storage/lmgr.h"
 #include "storage/proc.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "storage/standby.h"
 #include "utils/memdebug.h"
@@ -135,6 +136,33 @@ typedef struct SMgrSortArray
 	SMgrRelation srel;
 } SMgrSortArray;
 
+/*
+ * Helper struct for read stream object used in
+ * RelationCopyStorageUsingBuffer() function.
+ */
+struct copy_storage_using_buffer_read_stream_private
+{
+	BlockNumber blocknum;
+	BlockNumber last_block;
+};
+
+/*
+ * Callback function to get next block for read stream object used in
+ * RelationCopyStorageUsingBuffer() function.
+ */
+static BlockNumber
+copy_storage_using_buffer_read_stream_next_block(ReadStream *stream,
+												 void *callback_private_data,
+												 void *per_buffer_data)
+{
+	struct copy_storage_using_buffer_read_stream_private *p = callback_private_data;
+
+	if (p->blocknum < p->last_block)
+		return p->blocknum++;
+
+	return InvalidBlockNumber;
+}
+
 /* GUC variables */
 bool		zero_damaged_pages = false;
 int			bgwriter_lru_maxpages = 100;
@@ -4688,6 +4716,9 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 	PGIOAlignedBlock buf;
 	BufferAccessStrategy bstrategy_src;
 	BufferAccessStrategy bstrategy_dst;
+	struct copy_storage_using_buffer_read_stream_private p;
+	ReadStream *src_stream;
+	SMgrRelation src_smgr;
 
 	/*
 	 * In general, we want to write WAL whenever wal_level > 'minimal', but we
@@ -4716,19 +4747,31 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 	bstrategy_src = GetAccessStrategy(BAS_BULKREAD);
 	bstrategy_dst = GetAccessStrategy(BAS_BULKWRITE);
 
+	/* Initalize streaming read */
+	p.blocknum = 0;
+	p.last_block = nblocks;
+	src_smgr = smgropen(srclocator, INVALID_PROC_NUMBER);
+	src_stream = read_stream_begin_smgr_relation(READ_STREAM_FULL,
+												 bstrategy_src,
+												 src_smgr,
+												 permanent ? RELPERSISTENCE_PERMANENT : RELPERSISTENCE_UNLOGGED,
+												 forkNum,
+												 copy_storage_using_buffer_read_stream_next_block,
+												 &p,
+												 0);
+
 	/* Iterate over each block of the source relation file. */
 	for (blkno = 0; blkno < nblocks; blkno++)
 	{
 		CHECK_FOR_INTERRUPTS();
 
 		/* Read block from source relation. */
-		srcBuf = ReadBufferWithoutRelcache(srclocator, forkNum, blkno,
-										   RBM_NORMAL, bstrategy_src,
-										   permanent);
+		srcBuf = read_stream_next_buffer(src_stream, NULL);
 		LockBuffer(srcBuf, BUFFER_LOCK_SHARE);
 		srcPage = BufferGetPage(srcBuf);
 
-		dstBuf = ReadBufferWithoutRelcache(dstlocator, forkNum, blkno,
+		dstBuf = ReadBufferWithoutRelcache(dstlocator, forkNum,
+										   BufferGetBlockNumber(srcBuf),
 										   RBM_ZERO_AND_LOCK, bstrategy_dst,
 										   permanent);
 		dstPage = BufferGetPage(dstBuf);
@@ -4748,6 +4791,8 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 		UnlockReleaseBuffer(dstBuf);
 		UnlockReleaseBuffer(srcBuf);
 	}
+	Assert(read_stream_next_buffer(src_stream, NULL) == InvalidBuffer);
+	read_stream_end(src_stream);
 
 	FreeAccessStrategy(bstrategy_src);
 	FreeAccessStrategy(bstrategy_dst);
-- 
2.45.2



^ permalink  raw  reply  [nested|flat] 4+ messages in thread

* Re: Use read streams in CREATE DATABASE command when the strategy is wal_log
@ 2024-07-16 12:19  Noah Misch <[email protected]>
  parent: Nazir Bilal Yavuz <[email protected]>
  0 siblings, 1 reply; 4+ messages in thread

From: Noah Misch @ 2024-07-16 12:19 UTC (permalink / raw)
  To: Nazir Bilal Yavuz <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>

On Tue, Jul 16, 2024 at 02:11:20PM +0300, Nazir Bilal Yavuz wrote:
> On Fri, 12 Jul 2024 at 02:52, Noah Misch <[email protected]> wrote:
> > On Tue, Apr 16, 2024 at 02:12:19PM +0300, Nazir Bilal Yavuz wrote:
> > > --- a/src/backend/storage/aio/read_stream.c
> > > +++ b/src/backend/storage/aio/read_stream.c
> > > @@ -549,7 +549,7 @@ read_stream_begin_relation(int flags,
> > >       {
> > >               stream->ios[i].op.rel = rel;
> > >               stream->ios[i].op.smgr = RelationGetSmgr(rel);
> > > -             stream->ios[i].op.smgr_persistence = 0;
> > > +             stream->ios[i].op.smgr_persistence = rel->rd_rel->relpersistence;
> >
> > Does the following comment in ReadBuffersOperation need an update?
> >
> >         /*
> >          * The following members should be set by the caller.  If only smgr is
> >          * provided without rel, then smgr_persistence can be set to override the
> >          * default assumption of RELPERSISTENCE_PERMANENT.
> >          */
> 
> I believe it does not need to be updated but I renamed
> 'ReadBuffersOperation.smgr_persistence' as
> 'ReadBuffersOperation.persistence'. So, this comment is updated as
> well. I think that rename suits better because persistence does not
> need to come from smgr, it could come from relation, too. Do you think
> it is a good idea? If it is, does it need a separate commit?

The rename is good.  I think the comment implies "persistence" is unused when
rel!=NULL.  That implication is true before the patch but false after the
patch.

> > > @@ -4667,19 +4698,31 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
> >
> > >       /* Iterate over each block of the source relation file. */
> > >       for (blkno = 0; blkno < nblocks; blkno++)
> > >       {
> > >               CHECK_FOR_INTERRUPTS();
> > >
> > >               /* Read block from source relation. */
> > > -             srcBuf = ReadBufferWithoutRelcache(srclocator, forkNum, blkno,
> > > -                                                                                RBM_NORMAL, bstrategy_src,
> > > -                                                                                permanent);
> > > +             srcBuf = read_stream_next_buffer(src_stream, NULL);
> > >               LockBuffer(srcBuf, BUFFER_LOCK_SHARE);
> >
> > I think this should check for read_stream_next_buffer() returning
> > InvalidBuffer.  pg_prewarm doesn't, but the other callers do, and I think the
> > other callers are a better model.  LockBuffer() doesn't check the
> > InvalidBuffer case, so let's avoid the style of using a
> > read_stream_next_buffer() return value without checking.
> 
> There is an assert in the LockBuffer which checks for the
> InvalidBuffer. If that is not enough, we may add an if check for
> InvalidBuffer but what should we do in this case? It should not
> happen, so erroring out may be a good idea.

I like this style from read_stream_reset():

	while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
	{
		...
	}

That is, don't iterate over block numbers.  Drain the stream until empty.  If
the stream returns a number of blocks higher or lower than we expected, we
won't detect that, and that's okay.  It's not a strong preference, so I'm open
to arguments against that from you or others.  A counterargument could be that
read_stream_reset() doesn't know the buffer count, so it has no choice.  The
counterargument could say that callers knowing the block count should use the
pg_prewarm() style, and others should use the read_stream_reset() style.






^ permalink  raw  reply  [nested|flat] 4+ messages in thread

* Re: Use read streams in CREATE DATABASE command when the strategy is wal_log
@ 2024-07-17 09:22  Nazir Bilal Yavuz <[email protected]>
  parent: Noah Misch <[email protected]>
  0 siblings, 0 replies; 4+ messages in thread

From: Nazir Bilal Yavuz @ 2024-07-17 09:22 UTC (permalink / raw)
  To: Noah Misch <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>

Hi,

On Tue, 16 Jul 2024 at 15:19, Noah Misch <[email protected]> wrote:
>
> On Tue, Jul 16, 2024 at 02:11:20PM +0300, Nazir Bilal Yavuz wrote:
> > On Fri, 12 Jul 2024 at 02:52, Noah Misch <[email protected]> wrote:
> > > On Tue, Apr 16, 2024 at 02:12:19PM +0300, Nazir Bilal Yavuz wrote:
> > > > --- a/src/backend/storage/aio/read_stream.c
> > > > +++ b/src/backend/storage/aio/read_stream.c
> > > > @@ -549,7 +549,7 @@ read_stream_begin_relation(int flags,
> > > >       {
> > > >               stream->ios[i].op.rel = rel;
> > > >               stream->ios[i].op.smgr = RelationGetSmgr(rel);
> > > > -             stream->ios[i].op.smgr_persistence = 0;
> > > > +             stream->ios[i].op.smgr_persistence = rel->rd_rel->relpersistence;
> > >
> > > Does the following comment in ReadBuffersOperation need an update?
> > >
> > >         /*
> > >          * The following members should be set by the caller.  If only smgr is
> > >          * provided without rel, then smgr_persistence can be set to override the
> > >          * default assumption of RELPERSISTENCE_PERMANENT.
> > >          */
> >
> > I believe it does not need to be updated but I renamed
> > 'ReadBuffersOperation.smgr_persistence' as
> > 'ReadBuffersOperation.persistence'. So, this comment is updated as
> > well. I think that rename suits better because persistence does not
> > need to come from smgr, it could come from relation, too. Do you think
> > it is a good idea? If it is, does it need a separate commit?
>
> The rename is good.  I think the comment implies "persistence" is unused when
> rel!=NULL.  That implication is true before the patch but false after the
> patch.

What makes it false after the patch? I think the logic did not change.
If there is rel, the value of persistence is obtained from
'rel->rd_rel->relpersistence'. If there is no rel, then smgr is used
to obtain its value.

> > > > @@ -4667,19 +4698,31 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
> > >
> > > >       /* Iterate over each block of the source relation file. */
> > > >       for (blkno = 0; blkno < nblocks; blkno++)
> > > >       {
> > > >               CHECK_FOR_INTERRUPTS();
> > > >
> > > >               /* Read block from source relation. */
> > > > -             srcBuf = ReadBufferWithoutRelcache(srclocator, forkNum, blkno,
> > > > -                                                                                RBM_NORMAL, bstrategy_src,
> > > > -                                                                                permanent);
> > > > +             srcBuf = read_stream_next_buffer(src_stream, NULL);
> > > >               LockBuffer(srcBuf, BUFFER_LOCK_SHARE);
> > >
> > > I think this should check for read_stream_next_buffer() returning
> > > InvalidBuffer.  pg_prewarm doesn't, but the other callers do, and I think the
> > > other callers are a better model.  LockBuffer() doesn't check the
> > > InvalidBuffer case, so let's avoid the style of using a
> > > read_stream_next_buffer() return value without checking.
> >
> > There is an assert in the LockBuffer which checks for the
> > InvalidBuffer. If that is not enough, we may add an if check for
> > InvalidBuffer but what should we do in this case? It should not
> > happen, so erroring out may be a good idea.
>
> I like this style from read_stream_reset():
>
>         while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
>         {
>                 ...
>         }
>
> That is, don't iterate over block numbers.  Drain the stream until empty.  If
> the stream returns a number of blocks higher or lower than we expected, we
> won't detect that, and that's okay.  It's not a strong preference, so I'm open
> to arguments against that from you or others.  A counterargument could be that
> read_stream_reset() doesn't know the buffer count, so it has no choice.  The
> counterargument could say that callers knowing the block count should use the
> pg_prewarm() style, and others should use the read_stream_reset() style.

I think what you said in the counter argument makes sense. Also, there
is an 'Assert(read_stream_next_buffer(src_stream, NULL) ==
InvalidBuffer);' after the loop. Which means all the blocks in the
stream are read and there is no block left.

v3 is attached. The only change is 'read_stream.c' changes in the 0003
are moved to 0002.

--
Regards,
Nazir Bilal Yavuz
Microsoft


Attachments:

  [text/x-patch] v3-0001-Refactor-PinBufferForBlock-to-remove-if-checks-ab.patch (4.8K, ../../CAN55FZ33iZ3LmfqGhGH9YU8-b5SQisD_KJmmtFAEsUR0=XL-gg@mail.gmail.com/2-v3-0001-Refactor-PinBufferForBlock-to-remove-if-checks-ab.patch)
  download | inline diff:
From 977cea9451e602d8d3d5e4f0cf3cd7dfea11879e Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Sun, 7 Apr 2024 22:33:36 +0300
Subject: [PATCH v3 1/3] Refactor PinBufferForBlock() to remove if checks about
 persistence

There are if checks in PinBufferForBlock() function to set persistence
of the relation and this function is called for the each block in the
relation. Instead of that, set persistence of the relation before
PinBufferForBlock() function.
---
 src/include/storage/bufmgr.h          |  4 ++--
 src/backend/storage/aio/read_stream.c |  2 +-
 src/backend/storage/buffer/bufmgr.c   | 32 +++++++++++++--------------
 3 files changed, 18 insertions(+), 20 deletions(-)

diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index a1e71013d32..ba9a1a289a9 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -116,12 +116,12 @@ struct ReadBuffersOperation
 {
 	/*
 	 * The following members should be set by the caller.  If only smgr is
-	 * provided without rel, then smgr_persistence can be set to override the
+	 * provided without rel, then persistence can be set to override the
 	 * default assumption of RELPERSISTENCE_PERMANENT.
 	 */
 	Relation	rel;
 	struct SMgrRelationData *smgr;
-	char		smgr_persistence;
+	char		persistence;
 	ForkNumber	forknum;
 	BufferAccessStrategy strategy;
 
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 74b9bae6313..58221649f27 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -551,7 +551,7 @@ read_stream_begin_relation(int flags,
 	{
 		stream->ios[i].op.rel = rel;
 		stream->ios[i].op.smgr = RelationGetSmgr(rel);
-		stream->ios[i].op.smgr_persistence = 0;
+		stream->ios[i].op.persistence = rel->rd_rel->relpersistence;
 		stream->ios[i].op.forknum = forknum;
 		stream->ios[i].op.strategy = strategy;
 	}
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 61816730955..7cbcdc63a6f 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -1104,7 +1104,7 @@ ZeroAndLockBuffer(Buffer buffer, ReadBufferMode mode, bool already_valid)
 static pg_attribute_always_inline Buffer
 PinBufferForBlock(Relation rel,
 				  SMgrRelation smgr,
-				  char smgr_persistence,
+				  char persistence,
 				  ForkNumber forkNum,
 				  BlockNumber blockNum,
 				  BufferAccessStrategy strategy,
@@ -1113,22 +1113,12 @@ PinBufferForBlock(Relation rel,
 	BufferDesc *bufHdr;
 	IOContext	io_context;
 	IOObject	io_object;
-	char		persistence;
 
 	Assert(blockNum != P_NEW);
 
-	/*
-	 * If there is no Relation it usually implies recovery and thus permanent,
-	 * but we take an argument because CreateAndCopyRelationData can reach us
-	 * with only an SMgrRelation for an unlogged relation that we don't want
-	 * to flag with BM_PERMANENT.
-	 */
-	if (rel)
-		persistence = rel->rd_rel->relpersistence;
-	else if (smgr_persistence == 0)
-		persistence = RELPERSISTENCE_PERMANENT;
-	else
-		persistence = smgr_persistence;
+	Assert((persistence == RELPERSISTENCE_TEMP ||
+			persistence == RELPERSISTENCE_PERMANENT ||
+			persistence == RELPERSISTENCE_UNLOGGED));
 
 	if (persistence == RELPERSISTENCE_TEMP)
 	{
@@ -1203,6 +1193,7 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 	ReadBuffersOperation operation;
 	Buffer		buffer;
 	int			flags;
+	char		persistence;
 
 	/*
 	 * Backward compatibility path, most code should use ExtendBufferedRel()
@@ -1224,12 +1215,19 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 		return ExtendBufferedRel(BMR_REL(rel), forkNum, strategy, flags);
 	}
 
+	if (rel)
+		persistence = rel->rd_rel->relpersistence;
+	else if (smgr_persistence == 0)
+		persistence = RELPERSISTENCE_PERMANENT;
+	else
+		persistence = smgr_persistence;
+
 	if (unlikely(mode == RBM_ZERO_AND_CLEANUP_LOCK ||
 				 mode == RBM_ZERO_AND_LOCK))
 	{
 		bool		found;
 
-		buffer = PinBufferForBlock(rel, smgr, smgr_persistence,
+		buffer = PinBufferForBlock(rel, smgr, persistence,
 								   forkNum, blockNum, strategy, &found);
 		ZeroAndLockBuffer(buffer, mode, found);
 		return buffer;
@@ -1241,7 +1239,7 @@ ReadBuffer_common(Relation rel, SMgrRelation smgr, char smgr_persistence,
 		flags = 0;
 	operation.smgr = smgr;
 	operation.rel = rel;
-	operation.smgr_persistence = smgr_persistence;
+	operation.persistence = persistence;
 	operation.forknum = forkNum;
 	operation.strategy = strategy;
 	if (StartReadBuffer(&operation,
@@ -1272,7 +1270,7 @@ StartReadBuffersImpl(ReadBuffersOperation *operation,
 
 		buffers[i] = PinBufferForBlock(operation->rel,
 									   operation->smgr,
-									   operation->smgr_persistence,
+									   operation->persistence,
 									   operation->forknum,
 									   blockNum + i,
 									   operation->strategy,
-- 
2.45.2



  [text/x-patch] v3-0002-Add-a-way-to-create-read-stream-object-by-using-S.patch (5.7K, ../../CAN55FZ33iZ3LmfqGhGH9YU8-b5SQisD_KJmmtFAEsUR0=XL-gg@mail.gmail.com/3-v3-0002-Add-a-way-to-create-read-stream-object-by-using-S.patch)
  download | inline diff:
From 408a9487f64397739f23839754000b28883d8ab6 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Wed, 17 Jul 2024 11:30:51 +0300
Subject: [PATCH v3 2/3] Add a way to create read stream object by using
 SMgrRelation

Currently read stream object can be created only by using Relation,
there is no way to create it by using SMgrRelation.

To achieve that, read_stream_begin_impl() function is created and
contents of the read_stream_begin_relation() is moved into this
function.

Both read_stream_begin_relation() and read_stream_begin_smgr_relation()
calls read_stream_begin_impl() by Relation and SMgrRelation
respectively.
---
 src/include/storage/read_stream.h     | 10 ++++
 src/backend/storage/aio/read_stream.c | 83 ++++++++++++++++++++++-----
 2 files changed, 79 insertions(+), 14 deletions(-)

diff --git a/src/include/storage/read_stream.h b/src/include/storage/read_stream.h
index f676d2cc20a..4e599904f26 100644
--- a/src/include/storage/read_stream.h
+++ b/src/include/storage/read_stream.h
@@ -15,6 +15,7 @@
 #define READ_STREAM_H
 
 #include "storage/bufmgr.h"
+#include "storage/smgr.h"
 
 /* Default tuning, reasonable for many users. */
 #define READ_STREAM_DEFAULT 0x00
@@ -57,6 +58,15 @@ extern ReadStream *read_stream_begin_relation(int flags,
 											  void *callback_private_data,
 											  size_t per_buffer_data_size);
 extern Buffer read_stream_next_buffer(ReadStream *stream, void **per_buffer_data);
+extern ReadStream *read_stream_begin_smgr_relation(int flags,
+												   BufferAccessStrategy strategy,
+												   SMgrRelation smgr,
+												   char smgr_persistence,
+												   ForkNumber forknum,
+												   ReadStreamBlockNumberCB callback,
+												   void *callback_private_data,
+												   size_t per_buffer_data_size);
+extern Buffer read_stream_next_buffer(ReadStream *stream, void **per_buffer_private);
 extern void read_stream_reset(ReadStream *stream);
 extern void read_stream_end(ReadStream *stream);
 
diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c
index 58221649f27..bd50c55788e 100644
--- a/src/backend/storage/aio/read_stream.c
+++ b/src/backend/storage/aio/read_stream.c
@@ -406,14 +406,16 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
  * write extra data for each block into the space provided to it.  It will
  * also receive callback_private_data for its own purposes.
  */
-ReadStream *
-read_stream_begin_relation(int flags,
-						   BufferAccessStrategy strategy,
-						   Relation rel,
-						   ForkNumber forknum,
-						   ReadStreamBlockNumberCB callback,
-						   void *callback_private_data,
-						   size_t per_buffer_data_size)
+static ReadStream *
+read_stream_begin_impl(int flags,
+					   BufferAccessStrategy strategy,
+					   Relation rel,
+					   SMgrRelation smgr,
+					   char persistence,
+					   ForkNumber forknum,
+					   ReadStreamBlockNumberCB callback,
+					   void *callback_private_data,
+					   size_t per_buffer_data_size)
 {
 	ReadStream *stream;
 	size_t		size;
@@ -422,9 +424,6 @@ read_stream_begin_relation(int flags,
 	int			strategy_pin_limit;
 	uint32		max_pinned_buffers;
 	Oid			tablespace_id;
-	SMgrRelation smgr;
-
-	smgr = RelationGetSmgr(rel);
 
 	/*
 	 * Decide how many I/Os we will allow to run at the same time.  That
@@ -434,7 +433,7 @@ read_stream_begin_relation(int flags,
 	 */
 	tablespace_id = smgr->smgr_rlocator.locator.spcOid;
 	if (!OidIsValid(MyDatabaseId) ||
-		IsCatalogRelation(rel) ||
+		(rel && IsCatalogRelation(rel)) ||
 		IsCatalogRelationOid(smgr->smgr_rlocator.locator.relNumber))
 	{
 		/*
@@ -550,8 +549,8 @@ read_stream_begin_relation(int flags,
 	for (int i = 0; i < max_ios; ++i)
 	{
 		stream->ios[i].op.rel = rel;
-		stream->ios[i].op.smgr = RelationGetSmgr(rel);
-		stream->ios[i].op.persistence = rel->rd_rel->relpersistence;
+		stream->ios[i].op.smgr = smgr;
+		stream->ios[i].op.persistence = persistence;
 		stream->ios[i].op.forknum = forknum;
 		stream->ios[i].op.strategy = strategy;
 	}
@@ -559,6 +558,62 @@ read_stream_begin_relation(int flags,
 	return stream;
 }
 
+/*
+ * Create a new read stream for reading a relation.
+ * See read_stream_begin_impl() for the detailed explanation.
+ */
+ReadStream *
+read_stream_begin_relation(int flags,
+						   BufferAccessStrategy strategy,
+						   Relation rel,
+						   ForkNumber forknum,
+						   ReadStreamBlockNumberCB callback,
+						   void *callback_private_data,
+						   size_t per_buffer_data_size)
+{
+	return read_stream_begin_impl(flags,
+								  strategy,
+								  rel,
+								  RelationGetSmgr(rel),
+								  rel->rd_rel->relpersistence,
+								  forknum,
+								  callback,
+								  callback_private_data,
+								  per_buffer_data_size);
+}
+
+/*
+ * Create a new read stream for reading a SMgr relation.
+ * See read_stream_begin_impl() for the detailed explanation.
+ */
+ReadStream *
+read_stream_begin_smgr_relation(int flags,
+								BufferAccessStrategy strategy,
+								SMgrRelation smgr,
+								char smgr_persistence,
+								ForkNumber forknum,
+								ReadStreamBlockNumberCB callback,
+								void *callback_private_data,
+								size_t per_buffer_data_size)
+{
+	char		persistence;
+
+	if (smgr_persistence == 0)
+		persistence = RELPERSISTENCE_PERMANENT;
+	else
+		persistence = smgr_persistence;
+
+	return read_stream_begin_impl(flags,
+								  strategy,
+								  NULL,
+								  smgr,
+								  persistence,
+								  forknum,
+								  callback,
+								  callback_private_data,
+								  per_buffer_data_size);
+}
+
 /*
  * Pull one pinned buffer out of a stream.  Each call returns successive
  * blocks in the order specified by the callback.  If per_buffer_data_size was
-- 
2.45.2



  [text/x-patch] v3-0003-Use-read-streams-in-CREATE-DATABASE-when-strategy.patch (4.0K, ../../CAN55FZ33iZ3LmfqGhGH9YU8-b5SQisD_KJmmtFAEsUR0=XL-gg@mail.gmail.com/4-v3-0003-Use-read-streams-in-CREATE-DATABASE-when-strategy.patch)
  download | inline diff:
From 2aee4cc322bf47f7eb82ad48a472d4c9f2706f1d Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <[email protected]>
Date: Tue, 16 Jul 2024 18:11:21 +0300
Subject: [PATCH v3 3/3] Use read streams in CREATE DATABASE when strategy is
 wal_log

CREATE DABASE command uses RelationCopyStorageUsingBuffer() function to
copy buffers when the strategy is wal_log. This function reads source
buffers then copies them to the destination buffers. Read streams are
used only only while reading source buffers because the destination
buffer is read by 'RBM_ZERO_AND_LOCK' option, so it is not important.
---
 src/backend/storage/buffer/bufmgr.c | 53 ++++++++++++++++++++++++++---
 1 file changed, 49 insertions(+), 4 deletions(-)

diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 7cbcdc63a6f..2acc109c4db 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -54,6 +54,7 @@
 #include "storage/ipc.h"
 #include "storage/lmgr.h"
 #include "storage/proc.h"
+#include "storage/read_stream.h"
 #include "storage/smgr.h"
 #include "storage/standby.h"
 #include "utils/memdebug.h"
@@ -135,6 +136,33 @@ typedef struct SMgrSortArray
 	SMgrRelation srel;
 } SMgrSortArray;
 
+/*
+ * Helper struct for read stream object used in
+ * RelationCopyStorageUsingBuffer() function.
+ */
+struct copy_storage_using_buffer_read_stream_private
+{
+	BlockNumber blocknum;
+	BlockNumber last_block;
+};
+
+/*
+ * Callback function to get next block for read stream object used in
+ * RelationCopyStorageUsingBuffer() function.
+ */
+static BlockNumber
+copy_storage_using_buffer_read_stream_next_block(ReadStream *stream,
+												 void *callback_private_data,
+												 void *per_buffer_data)
+{
+	struct copy_storage_using_buffer_read_stream_private *p = callback_private_data;
+
+	if (p->blocknum < p->last_block)
+		return p->blocknum++;
+
+	return InvalidBlockNumber;
+}
+
 /* GUC variables */
 bool		zero_damaged_pages = false;
 int			bgwriter_lru_maxpages = 100;
@@ -4688,6 +4716,9 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 	PGIOAlignedBlock buf;
 	BufferAccessStrategy bstrategy_src;
 	BufferAccessStrategy bstrategy_dst;
+	struct copy_storage_using_buffer_read_stream_private p;
+	ReadStream *src_stream;
+	SMgrRelation src_smgr;
 
 	/*
 	 * In general, we want to write WAL whenever wal_level > 'minimal', but we
@@ -4716,19 +4747,31 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 	bstrategy_src = GetAccessStrategy(BAS_BULKREAD);
 	bstrategy_dst = GetAccessStrategy(BAS_BULKWRITE);
 
+	/* Initalize streaming read */
+	p.blocknum = 0;
+	p.last_block = nblocks;
+	src_smgr = smgropen(srclocator, INVALID_PROC_NUMBER);
+	src_stream = read_stream_begin_smgr_relation(READ_STREAM_FULL,
+												 bstrategy_src,
+												 src_smgr,
+												 permanent ? RELPERSISTENCE_PERMANENT : RELPERSISTENCE_UNLOGGED,
+												 forkNum,
+												 copy_storage_using_buffer_read_stream_next_block,
+												 &p,
+												 0);
+
 	/* Iterate over each block of the source relation file. */
 	for (blkno = 0; blkno < nblocks; blkno++)
 	{
 		CHECK_FOR_INTERRUPTS();
 
 		/* Read block from source relation. */
-		srcBuf = ReadBufferWithoutRelcache(srclocator, forkNum, blkno,
-										   RBM_NORMAL, bstrategy_src,
-										   permanent);
+		srcBuf = read_stream_next_buffer(src_stream, NULL);
 		LockBuffer(srcBuf, BUFFER_LOCK_SHARE);
 		srcPage = BufferGetPage(srcBuf);
 
-		dstBuf = ReadBufferWithoutRelcache(dstlocator, forkNum, blkno,
+		dstBuf = ReadBufferWithoutRelcache(dstlocator, forkNum,
+										   BufferGetBlockNumber(srcBuf),
 										   RBM_ZERO_AND_LOCK, bstrategy_dst,
 										   permanent);
 		dstPage = BufferGetPage(dstBuf);
@@ -4748,6 +4791,8 @@ RelationCopyStorageUsingBuffer(RelFileLocator srclocator,
 		UnlockReleaseBuffer(dstBuf);
 		UnlockReleaseBuffer(srcBuf);
 	}
+	Assert(read_stream_next_buffer(src_stream, NULL) == InvalidBuffer);
+	read_stream_end(src_stream);
 
 	FreeAccessStrategy(bstrategy_src);
 	FreeAccessStrategy(bstrategy_dst);
-- 
2.45.2



^ permalink  raw  reply  [nested|flat] 4+ messages in thread


end of thread, other threads:[~2024-07-17 09:22 UTC | newest]

Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-03-12 20:43 [PATCH 7/8] add wal_compression_method=zstd Justin Pryzby <[email protected]>
2024-07-16 11:11 Re: Use read streams in CREATE DATABASE command when the strategy is wal_log Nazir Bilal Yavuz <[email protected]>
2024-07-16 12:19 ` Re: Use read streams in CREATE DATABASE command when the strategy is wal_log Noah Misch <[email protected]>
2024-07-17 09:22   ` Re: Use read streams in CREATE DATABASE command when the strategy is wal_log Nazir Bilal Yavuz <[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