readme

Author: aykut-bozkurt

Cargo.lock

@@ -28,6 +28,7 @@ aws-credential-types = {version = "1", default-features = false}
 azure_storage = {version = "0.21", default-features = false}
 futures = "0.3"
 home = "0.5"
+libc = {version = "0.2", default-features = false }
 object_store = {version = "=0.12.2", default-features = false, features = ["aws", "azure", "fs", "gcp", "http"]}
 once_cell = "1"
 parquet = {version = "55", default-features = false, features = [

Cargo.toml

@@ -18,7 +18,9 @@ COPY table FROM 's3://mybucket/data.parquet' WITH (format 'parquet');
 ## Quick Reference
 - [Installation From Source](#installation-from-source)
 - [Usage](#usage)
-  - [Copy FROM/TO Parquet files TO/FROM Postgres tables](#copy-tofrom-parquet-files-fromto-postgres-tables)
+  - [Copy FROM/TO Parquet files TO/FROM Postgres tables](#copy-fromto-parquet-files-tofrom-postgres-tables)
+  - [COPY FROM/TO Parquet stdin/stdout TO/FROM Postgres tables)](#copy-fromto-parquet-stdinstdout-tofrom-postgres-tables)
+  - [COPY FROM/TO Parquet program stream TO/FROM Postgres tables)](#copy-fromto-parquet-program-stream-tofrom-postgres-tables)
   - [Inspect Parquet schema](#inspect-parquet-schema)
   - [Inspect Parquet metadata](#inspect-parquet-metadata)
   - [Inspect Parquet column statistics](#inspect-parquet-column-statistics)
@@ -56,11 +58,11 @@ psql> "CREATE EXTENSION pg_parquet;"
 
 ## Usage
 There are mainly 3 things that you can do with `pg_parquet`:
-1. You can export Postgres tables/queries to Parquet files,
+1. You can export Postgres tables/queries to Parquet files, stdin/stdout or a program's stream,
 2. You can ingest data from Parquet files to Postgres tables,
 3. You can inspect the schema and metadata of Parquet files.
 
-### COPY to/from Parquet files from/to Postgres tables
+### COPY from/to Parquet files to/from Postgres tables
 You can use PostgreSQL's `COPY` command to read and write from/to Parquet files. Below is an example of how to write a PostgreSQL table, with complex types, into a Parquet file and then to read the Parquet file content back into the same table.
 
 ```sql
@@ -99,7 +101,9 @@ COPY product_example FROM '/tmp/product_example.parquet';
 SELECT * FROM product_example;
 ```
 
-You can also use `COPY` command to read and write Parquet stream from/to standard input and output. Below is an example usage (you have to specify `format = parquet`):
+### COPY from/to Parquet stdin/stdout to/from Postgres tables
+
+You can use `COPY` command to read and write Parquet stream from/to standard input and output. Below is an example usage (you have to specify `format = parquet`):
 
 ```bash
 psql -d pg_parquet -p 28817 -h localhost -c "create table product_example_reconstructed (like product_example);"
@@ -109,6 +113,19 @@ psql -d pg_parquet -p 28817 -h localhost -c "copy product_example to stdout (for
  COPY 2
 ```
 
+### COPY from/to Parquet program stream to/from Postgres tables
+
+You can use `COPY` command to read and write Parquet stream from/to a program's input and output. Below is an example usage (you have to specify `format = parquet`):
+
+```bash
+psql -d pg_parquet -p 28817 -h localhost -c "copy product_example_reconstructed to program 'cat > /tmp/test.parquet' (format parquet);"
+ COPY 2
+
+psql -d pg_parquet -p 28817 -h localhost -c "copy product_example_reconstructed from program 'cat /tmp/test.parquet' (format parquet);"
+ COPY 2
+```
+
+
 ### Inspect Parquet schema
 You can call `SELECT * FROM parquet.schema(<uri>)` to discover the schema of the Parquet file at given uri.
 

README.md

@@ -17,7 +17,7 @@ use pgrx::{
     pg_sys::{
         get_role_oid, has_privs_of_role, palloc0, superuser, AsPgCStr, ClosePipeStream, DataDir,
         FileClose, FilePathName, GetUserId, InvalidOid, OpenPipeStream, OpenTemporaryFile,
-        TempTablespacePath, __sFILE, MAXPGPATH, PG_BINARY_R, PG_BINARY_W,
+        TempTablespacePath, MAXPGPATH, PG_BINARY_R, PG_BINARY_W,
     },
 };
 use url::Url;
@@ -45,7 +45,7 @@ pub(crate) struct ParsedUriInfo {
     // pipe_file is used to hold the pipe file descriptor for copying data to/from a program
     // call open_program_pipe to open the pipe to a program
     pub(crate) is_program: bool,
-    pub(crate) pipe_file: *mut __sFILE,
+    pub(crate) pipe_file: *mut libc::FILE,
 }
 
 impl ParsedUriInfo {
@@ -68,11 +68,15 @@ impl ParsedUriInfo {
             panic!("Failed to open pipe stream for program: {}", program);
         }
 
-        self.pipe_file = pipe_file;
+        self.pipe_file = pipe_file as _;
 
-        let pipe_fd = (unsafe { *self.pipe_file })._file;
+        let fd = unsafe { libc::fileno(self.pipe_file) };
 
-        unsafe { File::from_raw_fd(pipe_fd as _) }
+        if fd < 0 {
+            panic!("Failed to get file descriptor for pipe stream: {}", program);
+        }
+
+        unsafe { File::from_raw_fd(fd) }
     }
 
     fn with_tmp_file() -> Self {
@@ -180,7 +184,7 @@ impl Drop for ParsedUriInfo {
 
         if !self.pipe_file.is_null() {
             // close pipe file, postgres api will remove it on close
-            unsafe { ClosePipeStream(self.pipe_file) };
+            unsafe { ClosePipeStream(self.pipe_file as _) };
         }
     }
 }