feat: retriever parquet

README.md

@@ -81,10 +81,12 @@ against a previous JSONL baseline.
 manifest-based collections of compressed JSONL fragments. It supports
 PostgreSQL and Neo4j, uncompressed, gzip, and zstd fragments, bounded keyset
 scans, resumable dump checkpoints, checksum validation before load, optional
-deterministic property scrubbing, and a read-throughput benchmark mode. It can
-also package dumps as single HPKE/ML-KEM encrypted TAR archives.
-See [cmd/retriever/README.md](cmd/retriever/README.md) for dump, encrypted
-archive, load, scrubbed dump, metrics verification, and benchmark examples.
+deterministic property scrubbing, optional write-only Parquet sidecars for
+analytical readers, and a read-throughput benchmark mode. It can also package
+dumps as single HPKE/ML-KEM encrypted TAR archives. See
+[cmd/retriever/README.md](cmd/retriever/README.md) for dump, Parquet sidecar,
+encrypted archive, load, scrubbed dump, metrics verification, and benchmark
+examples.
 The same import/export functionality is available to library consumers from
 `github.com/specterops/dawgs/retriever`; callers provide an already-open
 `graph.Database`, and archive helpers support both path-based and stream-based

cmd/retriever/README.md

@@ -37,6 +37,34 @@ Use repeated `-graph` flags to dump multiple named graphs. For PostgreSQL,
 validates that expected node and edge partitions exist. For Neo4j, `-all-graphs`
 means the selected Neo4j database only.
 
+Pass `-parquet` to write optional Parquet sidecars alongside the JSONL
+fragments:
+
+```bash
+retriever dump \
+  -connection "$CONNECTION_STRING" \
+  -out ./dumpdir \
+  -graph default \
+  -scrub none \
+  -parquet
+```
+
+For example, `graphs/default/nodes-000001.jsonl.zst` is paired with
+`graphs/default/nodes-000001.parquet`, and
+`graphs/default/edges-000001.jsonl.zst` with
+`graphs/default/edges-000001.parquet`. Node sidecar rows contain `id` (string),
+`kinds` (list of strings), and `properties` (`VARIANT`). Edge sidecar rows
+contain `start_id`, `end_id`, and `kind` (strings), plus `properties`
+(`VARIANT`). The `VARIANT` property preserves JSON-like scalar, object, array,
+and null values for analytical readers.
+
+Parquet sidecars are optional output and are not part of the production
+collection path: loading, collection verification, manifest checksums and
+accounting, and encrypted archives remain JSONL-only. Interrupted Parquet
+dumps resume from committed JSONL/Parquet shard pairs. For each checkpointed
+JSONL fragment, resume requires the Parquet partner to be present as a regular
+file; it does not reopen or validate the sidecar's Parquet contents.
+
 Existing non-empty output directories are refused unless `-force` is supplied.
 The manifest is written last as `manifest.json`; if a dump fails before that
 point, the directory is intentionally left for inspection without a success

cmd/retriever/main.go

@@ -91,6 +91,7 @@ func (s commandRuntime) runDump(ctx context.Context, args []string) error {
 	flags.StringVar(&cfg.OutputDir, "out", "", "Output collection directory.")
 	flags.BoolVar(&cfg.Force, "force", false, "Replace an existing non-empty output directory.")
 	flags.BoolVar(&cfg.Resume, "resume", false, "Resume an interrupted dump from its validated checkpoint.")
+	flags.BoolVar(&cfg.Parquet, "parquet", false, "Also write Parquet sidecars for each JSONL fragment.")
 	flags.StringVar(&archiveOut, "archive-out", "", "Optional encrypted archive output path.")
 	flags.StringVar(&recipientPath, "recipient", "", "Recipient public key for -archive-out.")
 	flags.StringVar(&scrubValue, "scrub", string(cfg.Scrub), "Scrub mode: none or full.")

cmd/retriever/main_test.go

@@ -33,7 +33,7 @@ func TestCommandRuntimeHelpAndValidation(t *testing.T) {
 		t.Fatalf("expected unknown command error, got %v", err)
 	}
 
-	err = runtime.run(context.Background(), []string{"dump", "-out", t.TempDir(), "-scrub", "full"})
+	err = runtime.run(context.Background(), []string{"dump", "-out", t.TempDir(), "-parquet", "-scrub", "full"})
 	if err == nil || !strings.Contains(err.Error(), "-scrub full requires") {
 		t.Fatalf("expected scrub salt validation error, got %v", err)
 	}

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/klauspost/compress v1.19.0
 	github.com/neo4j/neo4j-go-driver/v5 v5.28.4
+	github.com/parquet-go/parquet-go v0.32.0
 	github.com/pashagolub/pgxmock/v5 v5.1.0
 	github.com/pelletier/go-toml/v2 v2.4.3
 	github.com/stretchr/testify v1.11.1
@@ -59,6 +60,7 @@ require (
 	github.com/alfatraining/structtag v1.0.0 // indirect
 	github.com/alingse/asasalint v0.0.11 // indirect
 	github.com/alingse/nilnesserr v0.2.0 // indirect
+	github.com/andybalholm/brotli v1.1.1 // indirect
 	github.com/ashanbrown/forbidigo/v2 v2.3.0 // indirect
 	github.com/ashanbrown/makezero/v2 v2.1.0 // indirect
 	github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
@@ -120,6 +122,7 @@ require (
 	github.com/golangci/swaggoswag v0.0.0-20250504205917-77f2aca3143e // indirect
 	github.com/golangci/unconvert v0.0.0-20250410112200-a129a6e6413e // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/gordonklaus/ineffassign v0.2.0 // indirect
 	github.com/gostaticanalysis/analysisutil v0.7.1 // indirect
 	github.com/gostaticanalysis/comment v1.5.0 // indirect
@@ -172,6 +175,9 @@ require (
 	github.com/nishanths/exhaustive v0.12.0 // indirect
 	github.com/nishanths/predeclared v0.2.2 // indirect
 	github.com/nunnatsa/ginkgolinter v0.23.0 // indirect
+	github.com/parquet-go/bitpack v1.0.0 // indirect
+	github.com/parquet-go/jsonlite v1.0.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.22.0 // indirect
@@ -212,6 +218,7 @@ require (
 	github.com/timonwong/loggercheck v0.11.0 // indirect
 	github.com/tomarrell/wrapcheck/v2 v2.12.0 // indirect
 	github.com/tommy-muehle/go-mnd/v2 v2.5.1 // indirect
+	github.com/twpayne/go-geom v1.6.1 // indirect
 	github.com/ultraware/funlen v0.2.0 // indirect
 	github.com/ultraware/whitespace v0.2.0 // indirect
 	github.com/uudashr/gocognit v1.2.1 // indirect

go.sum

@@ -60,6 +60,8 @@ github.com/alingse/asasalint v0.0.11 h1:SFwnQXJ49Kx/1GghOFz1XGqHYKp21Kq1nHad/0WQ
 github.com/alingse/asasalint v0.0.11/go.mod h1:nCaoMhw7a9kSJObvQyVzNTPBDbNpdocqrSP7t/cW5+I=
 github.com/alingse/nilnesserr v0.2.0 h1:raLem5KG7EFVb4UIDAXgrv3N2JIaffeKNtcEXkEWd/w=
 github.com/alingse/nilnesserr v0.2.0/go.mod h1:1xJPrXonEtX7wyTq8Dytns5P2hNzoWymVUIaKm4HNFg=
+github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
+github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
 github.com/ashanbrown/forbidigo/v2 v2.3.0 h1:OZZDOchCgsX5gvToVtEBoV2UWbFfI6RKQTir2UZzSxo=
@@ -443,10 +445,18 @@ github.com/otiai10/curr v0.0.0-20150429015615-9b4961190c95/go.mod h1:9qAhocn7zKJ
 github.com/otiai10/curr v1.0.0/go.mod h1:LskTG5wDwr8Rs+nNQ+1LlxRjAtTZZjtJW4rMXl6j4vs=
 github.com/otiai10/mint v1.3.0/go.mod h1:F5AjcsTsWUqX+Na9fpHb52P8pcRX2CI6A3ctIT91xUo=
 github.com/otiai10/mint v1.3.1/go.mod h1:/yxELlJQ0ufhjUwhshSj+wFjZ78CnZ48/1wtmBH1OTc=
+github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxPcQA=
+github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs=
+github.com/parquet-go/jsonlite v1.0.0 h1:87QNdi56wOfsE5bdgas0vRzHPxfJgzrXGml1zZdd7VU=
+github.com/parquet-go/jsonlite v1.0.0/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0=
+github.com/parquet-go/parquet-go v0.32.0 h1:NWDqTUHfrCS4cJP/Fj2HlxvqsrVedWG3sayMkf+znzM=
+github.com/parquet-go/parquet-go v0.32.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg=
 github.com/pashagolub/pgxmock/v5 v5.1.0 h1:NZ4pl82b335sEGIbD/+tk2fVIgVs3yNWr1R42ukpUvU=
 github.com/pashagolub/pgxmock/v5 v5.1.0/go.mod h1:8IJct22b7+EuqecVmYb9aKiENJLLqTsbjFHXH/znAEg=
 github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY=
 github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
+github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
+github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -568,6 +578,8 @@ github.com/tomarrell/wrapcheck/v2 v2.12.0 h1:H/qQ1aNWz/eeIhxKAFvkfIA+N7YDvq6TWVF
 github.com/tomarrell/wrapcheck/v2 v2.12.0/go.mod h1:AQhQuZd0p7b6rfW+vUwHm5OMCGgp63moQ9Qr/0BpIWo=
 github.com/tommy-muehle/go-mnd/v2 v2.5.1 h1:NowYhSdyE/1zwK9QCLeRb6USWdoif80Ie+v+yU8u1Zw=
 github.com/tommy-muehle/go-mnd/v2 v2.5.1/go.mod h1:WsUAkMJMYww6l/ufffCD3m+P7LEvr8TnZn9lwVDlgzw=
+github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4=
+github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028=
 github.com/ultraware/funlen v0.2.0 h1:gCHmCn+d2/1SemTdYMiKLAHFYxTYz7z9VIDRaTGyLkI=
 github.com/ultraware/funlen v0.2.0/go.mod h1:ZE0q4TsJ8T1SQcjmkhN/w+MceuatI6pBFSxxyteHIJA=
 github.com/ultraware/whitespace v0.2.0 h1:TYowo2m9Nfj1baEQBjuHzvMRbp19i+RCcRYrSWoFa+g=

retriever/archive_envelope.go

@@ -341,10 +341,6 @@ func UnpackEncryptedCollectionArchiveWithOptions(reader io.Reader, outputDir str
 	return nil
 }
 
-func unpackEncryptedCollectionArchiveToDirectory(reader io.Reader, outputDir string, identity hpke.PrivateKey) error {
-	return UnpackEncryptedCollectionArchive(reader, outputDir, identity)
-}
-
 func createUnpackStagingDirectory(outputDir string, force bool) (string, error) {
 	outputDir = strings.TrimSpace(outputDir)
 	if outputDir == "" {

retriever/archive_tar.go

@@ -376,11 +376,6 @@ func unpackTar(reader io.Reader, outputDir string, force bool) error {
 	return UnpackTar(reader, outputDir, force)
 }
 
-func unpackTarFile(reader io.Reader, outputDir, relativePath string, expectedSize int64) error {
-	_, err := unpackTarFileTracked(reader, outputDir, relativePath, expectedSize, false)
-	return err
-}
-
 func unpackTarFileTracked(reader io.Reader, outputDir, relativePath string, expectedSize int64, trackIntegrity bool) (unpackedFileIntegrity, error) {
 	absolutePath := filepath.Join(outputDir, filepath.FromSlash(relativePath))
 	if err := os.MkdirAll(filepath.Dir(absolutePath), 0o755); err != nil {

retriever/compression.go

@@ -156,11 +156,14 @@ func newDecompressionReader(reader io.Reader, codec CompressionCodec) (io.ReadCl
 }
 
 func newCompressedJSONLinesWriter(path string, codec CompressionCodec, zstdLevel int) (*compressedJSONLinesWriter, error) {
+	return newCompressedJSONLinesWriterAtPaths(path, path+".tmp", codec, zstdLevel)
+}
+
+func newCompressedJSONLinesWriterAtPaths(path, tempPath string, codec CompressionCodec, zstdLevel int) (*compressedJSONLinesWriter, error) {
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return nil, fmt.Errorf("create fragment directory: %w", err)
 	}
 
-	tempPath := path + ".tmp"
 	file, err := os.OpenFile(tempPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
 	if err != nil {
 		return nil, fmt.Errorf("open fragment temp file: %w", err)
@@ -215,6 +218,21 @@ func (s *compressedJSONLinesWriter) Count() int {
 }
 
 func (s *compressedJSONLinesWriter) Close() (FileManifest, error) {
+	fileEntry, err := s.finalize()
+	if err != nil {
+		return FileManifest{}, err
+	}
+
+	if err := os.Rename(s.tempPath, s.path); err != nil {
+		_ = os.Remove(s.tempPath)
+
+		return FileManifest{}, fmt.Errorf("rename fragment: %w", err)
+	}
+
+	return fileEntry, nil
+}
+
+func (s *compressedJSONLinesWriter) finalize() (FileManifest, error) {
 	if s.closed {
 		return FileManifest{}, fmt.Errorf("close JSONL fragment more than once")
 	}
@@ -233,12 +251,6 @@ func (s *compressedJSONLinesWriter) Close() (FileManifest, error) {
 		return FileManifest{}, fmt.Errorf("close fragment file: %w", err)
 	}
 
-	if err := os.Rename(s.tempPath, s.path); err != nil {
-		_ = os.Remove(s.tempPath)
-
-		return FileManifest{}, fmt.Errorf("rename fragment: %w", err)
-	}
-
 	return FileManifest{
 		Count:             s.count,
 		CompressedBytes:   s.compressedCounter.count,