From 78fc489d4739d0aee111fcab0ec6a7a489ab04e6 Mon Sep 17 00:00:00 2001 From: aykut-bozkurt <51649454+aykut-bozkurt@users.noreply.github.com> Date: Fri, 25 Oct 2024 23:20:01 +0300 Subject: [PATCH] Add pg{14,15} to support matrix (#57) 3 functions need to be ported to support PG{14,15}. pg_analyze_and_rewrite for pg14 pg_analyze_and_rewrite_fixedparams for >=pg15 Value for pg14 String for >= pg15 EmitWarningsOnPlaceholders for pg14 MarkGUCPrefixReserved for >= pg15 --- .github/workflows/ci.yml | 8 +- Cargo.toml | 6 +- README.md | 6 +- src/lib.rs | 4 +- src/parquet_copy_hook.rs | 1 + src/parquet_copy_hook/copy_from.rs | 20 +-- src/parquet_copy_hook/copy_to.rs | 20 +-- .../copy_to_dest_receiver.rs | 18 +- src/parquet_copy_hook/copy_utils.rs | 12 +- src/parquet_copy_hook/hook.rs | 154 +++++++++--------- src/parquet_copy_hook/pg_compat.rs | 67 ++++++++ src/type_compat/geometry.rs | 17 +- 12 files changed, 202 insertions(+), 131 deletions(-) create mode 100644 src/parquet_copy_hook/pg_compat.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0e8d3e6..403e54c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - postgres: [ 16, 17 ] + postgres: [ 14, 15, 16, 17 ] env: PG_MAJOR: ${{ matrix.postgres }} @@ -104,9 +104,9 @@ jobs: - name: Run tests run: | # Set up permissions so that the current user below can create extensions - sudo chmod a+rwx $(pg_config --pkglibdir) \ - $(pg_config --sharedir)/extension \ - /var/run/postgresql/ + sudo chmod a+rwx $(pg_config --pkglibdir) \ + $(pg_config --sharedir)/extension \ + /var/run/postgresql/ # pgrx tests with runas argument ignores environment variables, so # we read env vars from .env file in tests (https://github.com/pgcentralfoundation/pgrx/pull/1674) diff --git a/Cargo.toml b/Cargo.toml index 2e170fa..09b056f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,8 +13,10 @@ path = "./src/bin/pgrx_embed.rs" [features] default = ["pg17"] -pg16 = ["pgrx/pg16", "pgrx-tests/pg16" ] -pg17 = ["pgrx/pg17", "pgrx-tests/pg17" ] +pg17 = ["pgrx/pg17", "pgrx-tests/pg17"] +pg16 = ["pgrx/pg16", "pgrx-tests/pg16"] +pg15 = ["pgrx/pg15", "pgrx-tests/pg15"] +pg14 = ["pgrx/pg14", "pgrx-tests/pg14"] pg_test = [] [dependencies] diff --git a/README.md b/README.md index ae5b394..7c679a1 100644 --- a/README.md +++ b/README.md @@ -235,8 +235,10 @@ There is currently only one GUC parameter to enable/disable the `pg_parquet`: > Any type that does not have a corresponding Parquet type will be represented, as a fallback mechanism, as `BYTE_ARRAY` with `STRING` logical type. e.g. `enum` ## Postgres Support Matrix -`pg_parquet` is tested with the following PostgreSQL versions: +`pg_parquet` supports the following PostgreSQL versions: | PostgreSQL Major Version | Supported | |--------------------------|-----------| -| 17 | ✅ | +| 14 | ✅ | +| 15 | ✅ | | 16 | ✅ | +| 17 | ✅ | diff --git a/src/lib.rs b/src/lib.rs index ce61540..daaf057 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,5 @@ use parquet_copy_hook::hook::{init_parquet_copy_hook, ENABLE_PARQUET_COPY_HOOK}; -use pg_sys::{AsPgCStr, MarkGUCPrefixReserved}; +use parquet_copy_hook::pg_compat::MarkGUCPrefixReserved; use pgrx::{prelude::*, GucContext, GucFlags, GucRegistry}; mod arrow_parquet; @@ -29,7 +29,7 @@ pub extern "C" fn _PG_init() { GucFlags::default(), ); - unsafe { MarkGUCPrefixReserved("pg_parquet".as_pg_cstr()) }; + MarkGUCPrefixReserved("pg_parquet"); init_parquet_copy_hook(); } diff --git a/src/parquet_copy_hook.rs b/src/parquet_copy_hook.rs index 5dafd4e..2fb3002 100644 --- a/src/parquet_copy_hook.rs +++ b/src/parquet_copy_hook.rs @@ -3,3 +3,4 @@ pub(crate) mod copy_to; pub(crate) mod copy_to_dest_receiver; pub(crate) mod copy_utils; pub(crate) mod hook; +pub(crate) mod pg_compat; diff --git a/src/parquet_copy_hook/copy_from.rs b/src/parquet_copy_hook/copy_from.rs index f1fdab9..bf3a878 100644 --- a/src/parquet_copy_hook/copy_from.rs +++ b/src/parquet_copy_hook/copy_from.rs @@ -104,32 +104,32 @@ extern "C" fn copy_parquet_data_to_buffer( // 3. Calls the executor's CopyFrom function to read data from the parquet file and write // it to the copy buffer. pub(crate) fn execute_copy_from( - p_stmt: PgBox, + p_stmt: &PgBox, query_string: &CStr, - query_env: PgBox, + query_env: &PgBox, uri: Url, ) -> u64 { - let rel_oid = copy_stmt_relation_oid(&p_stmt); + let rel_oid = copy_stmt_relation_oid(p_stmt); copy_from_stmt_ensure_row_level_security(rel_oid); - let lock_mode = copy_stmt_lock_mode(&p_stmt); + let lock_mode = copy_stmt_lock_mode(p_stmt); let relation = unsafe { PgRelation::with_lock(rel_oid, lock_mode) }; - let p_state = copy_stmt_create_parse_state(query_string, &query_env); + let p_state = copy_stmt_create_parse_state(query_string, query_env); - let ns_item = copy_stmt_create_namespace_item(&p_stmt, &p_state, &relation); + let ns_item = copy_stmt_create_namespace_item(p_stmt, &p_state, &relation); - let mut where_clause = copy_from_stmt_where_clause(&p_stmt); + let mut where_clause = copy_from_stmt_where_clause(p_stmt); if !where_clause.is_null() { where_clause = copy_from_stmt_transform_where_clause(&p_state, &ns_item, where_clause); } - let attribute_list = copy_stmt_attribute_list(&p_stmt); + let attribute_list = copy_stmt_attribute_list(p_stmt); - let tupledesc = create_filtered_tupledesc_for_relation(&p_stmt, &relation); + let tupledesc = create_filtered_tupledesc_for_relation(p_stmt, &relation); unsafe { // parquet reader context is used throughout the COPY FROM operation. @@ -137,7 +137,7 @@ pub(crate) fn execute_copy_from( push_parquet_reader_context(parquet_reader_context); // makes sure to set binary format - let copy_options = copy_from_stmt_create_option_list(&p_stmt); + let copy_options = copy_from_stmt_create_option_list(p_stmt); let copy_from_state = BeginCopyFrom( p_state.as_ptr(), diff --git a/src/parquet_copy_hook/copy_to.rs b/src/parquet_copy_hook/copy_to.rs index 1e8df78..d6ed382 100644 --- a/src/parquet_copy_hook/copy_to.rs +++ b/src/parquet_copy_hook/copy_to.rs @@ -3,8 +3,8 @@ use std::ffi::{c_char, CStr}; use pgrx::{ ereport, is_a, pg_sys::{ - makeRangeVar, pg_analyze_and_rewrite_fixedparams, pg_plan_query, A_Star, ColumnRef, - CommandTag, CopyStmt, CreateNewPortal, DestReceiver, GetActiveSnapshot, + makeRangeVar, pg_plan_query, A_Star, ColumnRef, CommandTag, CopyStmt, CreateNewPortal, + DestReceiver, GetActiveSnapshot, Node, NodeTag::{self, T_CopyStmt}, ParamListInfoData, PlannedStmt, PortalDefineQuery, PortalDrop, PortalRun, PortalStart, QueryCompletion, QueryEnvironment, RawStmt, ResTarget, SelectStmt, CURSOR_OPT_PARALLEL_OK, @@ -14,8 +14,9 @@ use pgrx::{ AllocatedByRust, PgBox, PgList, PgLogLevel, PgRelation, PgSqlErrorCode, }; -use crate::parquet_copy_hook::copy_utils::{ - copy_stmt_has_relation, copy_stmt_lock_mode, copy_stmt_relation_oid, +use crate::parquet_copy_hook::{ + copy_utils::{copy_stmt_has_relation, copy_stmt_lock_mode, copy_stmt_relation_oid}, + pg_compat::pg_analyze_and_rewrite, }; // execute_copy_to_with_dest_receiver executes a COPY TO statement with our custom DestReceiver @@ -28,8 +29,8 @@ use crate::parquet_copy_hook::copy_utils::{ pub(crate) fn execute_copy_to_with_dest_receiver( p_stmt: &PgBox, query_string: &CStr, - params: PgBox, - query_env: PgBox, + params: &PgBox, + query_env: &PgBox, parquet_dest: PgBox, ) -> u64 { unsafe { @@ -51,11 +52,9 @@ pub(crate) fn execute_copy_to_with_dest_receiver( let raw_query = prepare_copy_to_raw_stmt(p_stmt, ©_stmt, &relation); - let rewritten_queries = pg_analyze_and_rewrite_fixedparams( + let rewritten_queries = pg_analyze_and_rewrite( raw_query.as_ptr(), query_string.as_ptr(), - std::ptr::null_mut(), - 0, query_env.as_ptr(), ); @@ -156,8 +155,7 @@ fn convert_copy_to_relation_to_select_stmt( target_list.push(target.into_pg()); } else { // SELECT a,b,... FROM relation - let attribute_name_list = - unsafe { PgList::::from_pg(copy_stmt.attlist) }; + let attribute_name_list = unsafe { PgList::::from_pg(copy_stmt.attlist) }; for attribute_name in attribute_name_list.iter_ptr() { let mut col_ref = unsafe { PgBox::::alloc_node(NodeTag::T_ColumnRef) }; diff --git a/src/parquet_copy_hook/copy_to_dest_receiver.rs b/src/parquet_copy_hook/copy_to_dest_receiver.rs index 07fe504..c0f6308 100644 --- a/src/parquet_copy_hook/copy_to_dest_receiver.rs +++ b/src/parquet_copy_hook/copy_to_dest_receiver.rs @@ -1,13 +1,13 @@ -use std::ffi::{c_char, CStr}; +use std::ffi::{c_char, CStr, CString}; use pg_sys::{ get_typlenbyval, slot_getallattrs, toast_raw_datum_size, AllocSetContextCreateExtended, - AsPgCStr, BlessTupleDesc, CommandDest, CurrentMemoryContext, Datum, DatumGetCString, - DestReceiver, HeapTupleData, List, MemoryContext, MemoryContextAllocZero, MemoryContextDelete, + AsPgCStr, BlessTupleDesc, CommandDest, CurrentMemoryContext, Datum, DestReceiver, + HeapTupleData, List, MemoryContext, MemoryContextAllocZero, MemoryContextDelete, MemoryContextReset, TupleDesc, TupleTableSlot, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, ALLOCSET_DEFAULT_MINSIZE, VARHDRSZ, }; -use pgrx::{prelude::*, PgList, PgMemoryContexts, PgTupleDesc}; +use pgrx::{prelude::*, FromDatum, PgList, PgMemoryContexts, PgTupleDesc}; use crate::{ arrow_parquet::{ @@ -373,15 +373,11 @@ fn tuple_column_sizes(tuple_datums: &[Option], tupledesc: &PgTupleDesc) - (unsafe { toast_raw_datum_size(*column_datum) }) as i32 - VARHDRSZ as i32 } else if typlen == -2 { // cstring - let cstring = unsafe { DatumGetCString(*column_datum) }; - let cstring = unsafe { - CStr::from_ptr(cstring) - .to_str() - .expect("cstring is not a valid CString") + CString::from_datum(*column_datum, false) + .expect("cannot get cstring from datum") }; - - cstring.len() as i32 + 1 + cstring.as_bytes().len() as i32 + 1 } else { // fixed size type typlen as i32 diff --git a/src/parquet_copy_hook/copy_utils.rs b/src/parquet_copy_hook/copy_utils.rs index cb21671..068e95e 100644 --- a/src/parquet_copy_hook/copy_utils.rs +++ b/src/parquet_copy_hook/copy_utils.rs @@ -5,7 +5,7 @@ use pgrx::{ pg_sys::{ addRangeTableEntryForRelation, defGetInt32, defGetInt64, defGetString, get_namespace_name, get_rel_namespace, makeDefElem, makeString, make_parsestate, quote_qualified_identifier, - AccessShareLock, AsPgCStr, CopyStmt, CreateTemplateTupleDesc, DefElem, List, NoLock, + AccessShareLock, AsPgCStr, CopyStmt, CreateTemplateTupleDesc, DefElem, List, NoLock, Node, NodeTag::T_CopyStmt, Oid, ParseNamespaceItem, ParseState, PlannedStmt, QueryEnvironment, RangeVar, RangeVarGetRelidExtended, RowExclusiveLock, TupleDescInitEntry, }, @@ -19,6 +19,8 @@ use crate::arrow_parquet::{ uri_utils::parse_uri, }; +use super::pg_compat::strVal; + pub(crate) fn validate_copy_to_options(p_stmt: &PgBox, uri: &Url) { validate_copy_option_names( p_stmt, @@ -444,13 +446,7 @@ fn copy_stmt_attribute_names(p_stmt: &PgBox) -> Vec { unsafe { PgList::from_pg(attribute_name_list) .iter_ptr() - .map(|attribute_name: *mut pgrx::pg_sys::String| { - let attribute_name = PgBox::from_pg(attribute_name); - CStr::from_ptr(attribute_name.sval) - .to_str() - .expect("cannot get attribute name for copy from statement") - .to_string() - }) + .map(|attribute_name: *mut Node| strVal(attribute_name)) .collect::>() } } diff --git a/src/parquet_copy_hook/hook.rs b/src/parquet_copy_hook/hook.rs index 1329817..f533a56 100644 --- a/src/parquet_copy_hook/hook.rs +++ b/src/parquet_copy_hook/hook.rs @@ -43,6 +43,72 @@ pub(crate) extern "C" fn init_parquet_copy_hook() { } } +fn process_copy_to_parquet( + p_stmt: &PgBox, + query_string: &CStr, + params: &PgBox, + query_env: &PgBox, +) -> u64 { + let uri = copy_stmt_uri(p_stmt).expect("uri is None"); + + let copy_from = false; + ensure_access_privilege_to_uri(&uri, copy_from); + + validate_copy_to_options(p_stmt, &uri); + + let row_group_size = copy_to_stmt_row_group_size(p_stmt); + let row_group_size_bytes = copy_to_stmt_row_group_size_bytes(p_stmt); + let compression = copy_to_stmt_compression(p_stmt, uri.clone()); + let compression_level = copy_to_stmt_compression_level(p_stmt, uri.clone()); + + PgTryBuilder::new(|| { + let parquet_dest = create_copy_to_parquet_dest_receiver( + uri_as_string(&uri).as_pg_cstr(), + &row_group_size, + &row_group_size_bytes, + &compression, + &compression_level.unwrap_or(INVALID_COMPRESSION_LEVEL), + ); + + let parquet_dest = unsafe { PgBox::from_pg(parquet_dest) }; + + execute_copy_to_with_dest_receiver(p_stmt, query_string, params, query_env, parquet_dest) + }) + .catch_others(|cause| { + // make sure to pop the parquet writer context + // In case we did not push the context, we should not throw an error while popping + let throw_error = false; + pop_parquet_writer_context(throw_error); + + cause.rethrow() + }) + .execute() +} + +fn process_copy_from_parquet( + p_stmt: &PgBox, + query_string: &CStr, + query_env: &PgBox, +) -> u64 { + let uri = copy_stmt_uri(p_stmt).expect("uri is None"); + + let copy_from = true; + ensure_access_privilege_to_uri(&uri, copy_from); + + validate_copy_from_options(p_stmt); + + PgTryBuilder::new(|| execute_copy_from(p_stmt, query_string, query_env, uri)) + .catch_others(|cause| { + // make sure to pop the parquet reader context + // In case we did not push the context, we should not throw an error while popping + let throw_error = false; + pop_parquet_reader_context(throw_error); + + cause.rethrow() + }) + .execute() +} + #[pg_guard] #[allow(clippy::too_many_arguments)] extern "C" fn parquet_copy_hook( @@ -59,85 +125,23 @@ extern "C" fn parquet_copy_hook( let query_string = unsafe { CStr::from_ptr(query_string) }; let params = unsafe { PgBox::from_pg(params) }; let query_env = unsafe { PgBox::from_pg(query_env) }; + let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) }; if ENABLE_PARQUET_COPY_HOOK.get() && is_copy_to_parquet_stmt(&p_stmt) { - let uri = copy_stmt_uri(&p_stmt).expect("uri is None"); - let copy_from = false; - - ensure_access_privilege_to_uri(&uri, copy_from); - - validate_copy_to_options(&p_stmt, &uri); - - let row_group_size = copy_to_stmt_row_group_size(&p_stmt); - let row_group_size_bytes = copy_to_stmt_row_group_size_bytes(&p_stmt); - let compression = copy_to_stmt_compression(&p_stmt, uri.clone()); - let compression_level = copy_to_stmt_compression_level(&p_stmt, uri.clone()); - - PgTryBuilder::new(|| { - let parquet_dest = create_copy_to_parquet_dest_receiver( - uri_as_string(&uri).as_pg_cstr(), - &row_group_size, - &row_group_size_bytes, - &compression, - &compression_level.unwrap_or(INVALID_COMPRESSION_LEVEL), - ); - - let parquet_dest = unsafe { PgBox::from_pg(parquet_dest) }; - - let nprocessed = execute_copy_to_with_dest_receiver( - &p_stmt, - query_string, - params, - query_env, - parquet_dest, - ); - - let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) }; - - if !completion_tag.is_null() { - completion_tag.nprocessed = nprocessed; - completion_tag.commandTag = CommandTag::CMDTAG_COPY; - } - }) - .catch_others(|cause| { - // make sure to pop the parquet writer context - // In case we did not push the context, we should not throw an error while popping - let throw_error = false; - pop_parquet_writer_context(throw_error); - - cause.rethrow() - }) - .execute(); + let nprocessed = process_copy_to_parquet(&p_stmt, query_string, ¶ms, &query_env); + if !completion_tag.is_null() { + completion_tag.nprocessed = nprocessed; + completion_tag.commandTag = CommandTag::CMDTAG_COPY; + } return; } else if ENABLE_PARQUET_COPY_HOOK.get() && is_copy_from_parquet_stmt(&p_stmt) { - let uri = copy_stmt_uri(&p_stmt).expect("uri is None"); - let copy_from = true; - - ensure_access_privilege_to_uri(&uri, copy_from); - - validate_copy_from_options(&p_stmt); - - PgTryBuilder::new(|| { - let nprocessed = execute_copy_from(p_stmt, query_string, query_env, uri); - - let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) }; - - if !completion_tag.is_null() { - completion_tag.nprocessed = nprocessed; - completion_tag.commandTag = CommandTag::CMDTAG_COPY; - } - }) - .catch_others(|cause| { - // make sure to pop the parquet reader context - // In case we did not push the context, we should not throw an error while popping - let throw_error = false; - pop_parquet_reader_context(throw_error); - - cause.rethrow() - }) - .execute(); + let nprocessed = process_copy_from_parquet(&p_stmt, query_string, &query_env); + if !completion_tag.is_null() { + completion_tag.nprocessed = nprocessed; + completion_tag.commandTag = CommandTag::CMDTAG_COPY; + } return; } @@ -151,7 +155,7 @@ extern "C" fn parquet_copy_hook( params.into_pg(), query_env.into_pg(), dest, - completion_tag, + completion_tag.into_pg(), ) } else { standard_ProcessUtility( @@ -162,7 +166,7 @@ extern "C" fn parquet_copy_hook( params.into_pg(), query_env.into_pg(), dest, - completion_tag, + completion_tag.into_pg(), ) } } diff --git a/src/parquet_copy_hook/pg_compat.rs b/src/parquet_copy_hook/pg_compat.rs new file mode 100644 index 0000000..1ff95eb --- /dev/null +++ b/src/parquet_copy_hook/pg_compat.rs @@ -0,0 +1,67 @@ +use std::ffi::{c_char, CStr}; + +use pgrx::pg_sys::{AsPgCStr, List, Node, QueryEnvironment, RawStmt}; + +pub(crate) fn pg_analyze_and_rewrite( + raw_stmt: *mut RawStmt, + query_string: *const c_char, + query_env: *mut QueryEnvironment, +) -> *mut List { + #[cfg(feature = "pg14")] + unsafe { + pgrx::pg_sys::pg_analyze_and_rewrite( + raw_stmt, + query_string, + std::ptr::null_mut(), + 0, + query_env, + ) + } + + #[cfg(any(feature = "pg15", feature = "pg16", feature = "pg17"))] + unsafe { + pgrx::pg_sys::pg_analyze_and_rewrite_fixedparams( + raw_stmt, + query_string, + std::ptr::null_mut(), + 0, + query_env, + ) + } +} + +#[allow(non_snake_case)] +pub(crate) fn strVal(val: *mut Node) -> String { + #[cfg(feature = "pg14")] + unsafe { + let val = (*(val as *mut pgrx::pg_sys::Value)).val.str_; + + CStr::from_ptr(val) + .to_str() + .expect("invalid string") + .to_string() + } + + #[cfg(any(feature = "pg15", feature = "pg16", feature = "pg17"))] + unsafe { + let val = (*(val as *mut pgrx::pg_sys::String)).sval; + + CStr::from_ptr(val) + .to_str() + .expect("invalid string") + .to_string() + } +} + +#[allow(non_snake_case)] +pub(crate) fn MarkGUCPrefixReserved(guc_prefix: &str) { + #[cfg(feature = "pg14")] + unsafe { + pgrx::pg_sys::EmitWarningsOnPlaceholders(guc_prefix.as_pg_cstr()) + } + + #[cfg(any(feature = "pg15", feature = "pg16", feature = "pg17"))] + unsafe { + pgrx::pg_sys::MarkGUCPrefixReserved(guc_prefix.as_pg_cstr()) + } +} diff --git a/src/type_compat/geometry.rs b/src/type_compat/geometry.rs index b8b2670..fbb3c25 100644 --- a/src/type_compat/geometry.rs +++ b/src/type_compat/geometry.rs @@ -4,11 +4,11 @@ use once_cell::sync::OnceCell; use pgrx::{ datum::UnboxDatum, pg_sys::{ - get_extension_oid, get_extension_schema, makeString, Anum_pg_type_oid, AsPgCStr, Datum, - GetSysCacheOid, InvalidOid, LookupFuncName, Oid, OidFunctionCall1Coll, - SysCacheIdentifier::TYPENAMENSP, BYTEAOID, + get_extension_oid, makeString, Anum_pg_type_oid, AsPgCStr, Datum, GetSysCacheOid, + InvalidOid, LookupFuncName, Oid, OidFunctionCall1Coll, SysCacheIdentifier::TYPENAMENSP, + BYTEAOID, }, - FromDatum, IntoDatum, PgList, + FromDatum, IntoDatum, PgList, Spi, }; // we need to reset the postgis context at each copy start @@ -56,8 +56,7 @@ impl PostgisContext { Some(postgis_ext_oid) }; - let postgis_ext_schema_oid = - postgis_ext_oid.map(|postgis_ext_oid| unsafe { get_extension_schema(postgis_ext_oid) }); + let postgis_ext_schema_oid = postgis_ext_oid.map(|_| Self::extension_schema_oid()); let st_asbinary_funcoid = postgis_ext_oid.map(|postgis_ext_oid| { Self::st_asbinary_funcoid( @@ -82,6 +81,12 @@ impl PostgisContext { } } + fn extension_schema_oid() -> Oid { + Spi::get_one("SELECT extnamespace FROM pg_extension WHERE extname = 'postgis'") + .expect("failed to get postgis extension schema") + .expect("postgis extension schema not found") + } + fn st_asbinary_funcoid(postgis_ext_oid: Oid, postgis_ext_schema_oid: Oid) -> Oid { unsafe { let postgis_geometry_typoid =