Repository: planetlabs/gpq (go)

Issue:
gpq convert output of Overture parquet files cannot be read by GDAL
I was testing the Overture maps data and realised it is only available in parquet and not geoparquet format. As I understand it this is a user case for gpq as mentioned in https://github.com/planetlabs/gpq/issues/57

The tools runs fine and seems to produce output, but I cannot read this using GDAL. Apologies if this is user error or should be a GDAL issue instead - please close if this is the case. 

Full steps to recreate below (note I was using gpq on a Windows machine, and testing the output on both Windows and Linux. 

Download data:

```
aws s3 cp --region us-west-2 --no-sign-request --recursive s3://overturemaps-us-west-2/release/2023-10-19-alpha.0/theme=buildings C:\Temp\buildings.parquet
```

Run conversion:

```ps1
$env:PATH += ";D:\Tools\gpq-windows-amd64"
gpq version
# 0.20.0

gpq convert part-00769-87dd7d19-acc8-4d4f-a5ba-20b407a79638.c000.zstd.parquet test.geo.parquet --from="parquet" --to="geoparquet"

# also tried without compression (no difference in terms of validity)

gpq convert part-00769-87dd7d19-acc8-4d4f-a5ba-20b407a79638.c000.zstd.parquet test.geo.parquet --from="parquet" --to="geoparquet" --compression="uncompressed"

gpq validate test.geo.parquet 

Summary: Passed 20 checks.

 ✓ file must include a "geo" metadata key
 ✓ metadata must be a JSON object
 ✓ metadata must include a "version" string
 ✓ metadata must include a "primary_column" string
 ✓ metadata must include a "columns" object
 ✓ column metadata must include the "primary_column" name
 ✓ column metadata must include a valid "encoding" string
 ✓ column metadata must include a "geometry_types" list
 ✓ optional "crs" must be null or a PROJJSON object
 ✓ optional "orientation" must be a valid string
 ✓ optional "edges" must be a valid string
 ✓ optional "bbox" must be an array of 4 or 6 numbers
 ✓ optional "epoch" must be a number
 ✓ geometry columns must not be grouped
 ✓ geometry columns must be stored using the BYTE_ARRAY parquet type
 ✓ geometry columns must be required or optional, not repeated
 ✓ all geometry values match the "encoding" metadata
 ✓ all geometry types must be included in the "geometry_types" metadata (if not empty)
 ✓ all polygon geometries must follow the "orientation" metadata (if present)
 ✓ all geometries must fall within the "bbox" metadata (if present)

```

QGIS opens the file but the attribute table is empty. Testing with `ogrinfo`:

```
ogrinfo --version
# GDAL 3.7.2, released 2023/09/05
ogrinfo test.geo.parquet

Warning 1: Field brand.names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field addresses of unhandled type list<element: struct<freeform: string, locality: string, postCode: string, region: string, country: string>> ignored
Warning 1: Field names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field sources of unhandled type list<element: struct<property: string, dataset: string, recordId: string, confidence: double>> ignored
INFO: Open of `test.geo.parquet'
      using driver `Parquet' successful.
1: test.geo
```

Trying to read the data gives the likely cause of the issue: `ERROR 1: ReadNext() failed: Malformed levels. min: 2 max: 2 out of range.  Max Level: 1`. 

```
ogrinfo test.geo.parquet -al

Warning 1: Field brand.names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field addresses of unhandled type list<element: struct<freeform: string, locality: string, postCode: string, region: string, country: string>> ignored
Warning 1: Field names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field sources of unhandled type list<element: struct<property: string, dataset: string, recordId: string, confidence: double>> ignored
INFO: Open of `test.geo.parquet'
      using driver `Parquet' successful.

Layer name: test.geo
Geometry: Unknown (any)
Feature Count: 815104
ERROR 1: ReadNext() failed: Malformed levels. min: 2 max: 2 out of range.  Max Level: 1
Layer SRS WKT:
GEOGCRS["WGS 84",
    ENSEMBLE["World Geodetic System 1984 ensemble",
        MEMBER["World Geodetic System 1984 (Transit)"],
        MEMBER["World Geodetic System 1984 (G730)"],
        MEMBER["World Geodetic System 1984 (G873)"],
        MEMBER["World Geodetic System 1984 (G1150)"],
        MEMBER["World Geodetic System 1984 (G1674)"],
        MEMBER["World Geodetic System 1984 (G1762)"],
        MEMBER["World Geodetic System 1984 (G2139)"],
        ELLIPSOID["WGS 84",6378137,298.257223563,
            LENGTHUNIT["metre",1]],
        ENSEMBLEACCURACY[2.0]],
    PRIMEM["Greenwich",0,
        ANGLEUNIT["degree",0.0174532925199433]],
    CS[ellipsoidal,2],
        AXIS["geodetic latitude (Lat)",north,
            ORDER[1],
            ANGLEUNIT["degree",0.0174532925199433]],
        AXIS["geodetic longitude (Lon)",east,
            ORDER[2],
            ANGLEUNIT["degree",0.0174532925199433]],
    USAGE[
        SCOPE["Horizontal component of 3D system."],
        AREA["World."],
        BBOX[-90,-180,90,180]],
    ID["EPSG",4326]]
Data axis to CRS axis mapping: 2,1
Geometry Column = geometry
categories.main: String (0.0)
categories.alternate: StringList (0.0)
level: Integer (0.0)
socials: StringList (0.0)
subType: String (0.0)
numFloors: Integer (0.0)
entityId: String (0.0)
class: String (0.0)
sourceTags: String(JSON) (0.0)
localityType: String (0.0)
emails: StringList (0.0)
drivingSide: String (0.0)
adminLevel: Integer (0.0)
road: String (0.0)
isoCountryCodeAlpha2: String (0.0)
isoSubCountryCode: String (0.0)
updateTime: String (0.0)
wikidata: String (0.0)
confidence: Real (0.0)
defaultLanguage: String (0.0)
brand.wikidata: String (0.0)
isIntermittent: Integer(Boolean) (0.0)
connectors: StringList (0.0)
surface: String (0.0)
version: Integer (0.0)
phones: StringList (0.0)
id: String (0.0)
context: String (0.0)
height: Real (0.0)
maritime: Integer(Boolean) (0.0)
websites: StringList (0.0)
isSalt: Integer(Boolean) (0.0)
bbox.minx: Real (0.0)
bbox.maxx: Real (0.0)
bbox.miny: Real (0.0)
bbox.maxy: Real (0.0)
ERROR 1: ReadNext() failed: Malformed levels. min: 2 max: 2 out of range.  Max Level: 1

```

Testing with the GDAL validate script from [here](https://raw.githubusercontent.com/OSGeo/gdal/master/swig/python/gdal-utils/osgeo_utils/samples/validate_geoparquet.py)

```

apt-get install python3-pip --fix-missing
python3 -m pip install jsonschema
python3 validate_geoparquet.py --check-data test.geo.parquet

Warning 1: Field brand.names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field brand.names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field addresses of unhandled type list<element: struct<freeform: string, locality: string, postCode: string, region: string, country: string>> ignored
Warning 1: Field names.common of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.official of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.alternate of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field names.short of unhandled type list<element: struct<value: string, language: string>> ignored
Warning 1: Field sources of unhandled type list<element: struct<property: string, dataset: string, recordId: string, confidence: double>> ignored
Segmentation fault
```

Pull request description:
Use the pqarrow file writer
This updates the Parquet transform-by-column code to use the `pqarrow.FileWriter` instead of a `file.Writer`.  When using the `file.Writer`, I'm seeing some cases where invalid Parquet is generated (see https://github.com/apache/arrow/issues/38503).

Fixes #102.

Patch:
diff --git a/internal/geoparquet/geoparquet.go b/internal/geoparquet/geoparquet.go
index 23f121c..4c98048 100644
--- a/internal/geoparquet/geoparquet.go
+++ b/internal/geoparquet/geoparquet.go
@@ -12,6 +12,7 @@ import (
 	"github.com/apache/arrow/go/v14/parquet"
 	"github.com/apache/arrow/go/v14/parquet/compress"
 	"github.com/apache/arrow/go/v14/parquet/file"
+	"github.com/apache/arrow/go/v14/parquet/pqarrow"
 	"github.com/apache/arrow/go/v14/parquet/schema"
 	"github.com/paulmach/orb/encoding/wkb"
 	"github.com/paulmach/orb/encoding/wkt"
@@ -154,7 +155,7 @@ func FromParquet(input parquet.ReaderAtSeeker, output io.Writer, convertOptions
 		return arrow.NewChunked(builder.Type(), transformed), nil
 	}
 
-	beforeClose := func(fileReader *file.Reader, fileWriter *file.Writer) error {
+	beforeClose := func(fileReader *file.Reader, fileWriter *pqarrow.FileWriter) error {
 		metadata := getMetadata(fileReader, convertOptions)
 		for name, geometryCol := range metadata.Columns {
 			if !datasetInfo.HasCollection(name) {
diff --git a/internal/pqutil/transform.go b/internal/pqutil/transform.go
index b96fd55..b0df118 100644
--- a/internal/pqutil/transform.go
+++ b/internal/pqutil/transform.go
@@ -26,7 +26,7 @@ type TransformConfig struct {
 	RowGroupLength  int
 	TransformSchema SchemaTransformer
 	TransformColumn ColumnTransformer
-	BeforeClose     func(*file.Reader, *file.Writer) error
+	BeforeClose     func(*file.Reader, *pqarrow.FileWriter) error
 }
 
 func getWriterProperties(config *TransformConfig, fileReader *file.Reader) (*parquet.WriterProperties, error) {
@@ -104,7 +104,16 @@ func TransformByColumn(config *TransformConfig) error {
 		return propErr
 	}
 
-	fileWriter := file.NewParquetWriter(config.Writer, outputSchema.Root(), file.WithWriterProps(writerProperties))
+	arrowSchema, arrowSchemaErr := pqarrow.FromParquet(outputSchema, &arrowReadProperties, fileReader.MetaData().KeyValueMetadata())
+	if arrowSchemaErr != nil {
+		return arrowSchemaErr
+	}
+
+	fileWriter, fileWriterErr := pqarrow.NewFileWriter(arrowSchema, config.Writer, writerProperties, pqarrow.DefaultWriterProps())
+	if fileWriterErr != nil {
+		return fileWriterErr
+	}
+
 	ctx := pqarrow.NewArrowWriteContext(context.Background(), nil)
 
 	if config.RowGroupLength > 0 {
@@ -120,7 +129,8 @@ func TransformByColumn(config *TransformConfig) error {
 		numRows := fileReader.NumRows()
 		numRowsWritten := int64(0)
 		for {
-			rowGroupWriter := fileWriter.AppendRowGroup()
+			fileWriter.NewRowGroup()
+			numRowsInGroup := 0
 			for fieldNum := 0; fieldNum < numFields; fieldNum += 1 {
 				colReader := columnReaders[fieldNum]
 				arr, readErr := colReader.NextBatch(int64(config.RowGroupLength))
@@ -139,18 +149,14 @@ func TransformByColumn(config *TransformConfig) error {
 					}
 					arr = transformed
 				}
-				colWriter, colWriterErr := pqarrow.NewArrowColumnWriter(arr, 0, int64(arr.Len()), outputManifest, rowGroupWriter, fieldNum)
-				if colWriterErr != nil {
-					return colWriterErr
+				if numRowsInGroup == 0 {
+					// TODO: propose fileWriter.RowGroupNumRows()
+					numRowsInGroup = arr.Len()
 				}
-				if err := colWriter.Write(ctx); err != nil {
+				if err := fileWriter.WriteColumnChunked(arr, 0, int64(arr.Len())); err != nil {
 					return err
 				}
 			}
-			numRowsInGroup, err := rowGroupWriter.NumRows()
-			if err != nil {
-				return err
-			}
 			numRowsWritten += int64(numRowsInGroup)
 			if numRowsWritten >= numRows {
 				break
@@ -160,7 +166,7 @@ func TransformByColumn(config *TransformConfig) error {
 		numRowGroups := fileReader.NumRowGroups()
 		for rowGroupIndex := 0; rowGroupIndex < numRowGroups; rowGroupIndex += 1 {
 			rowGroupReader := arrowReader.RowGroup(rowGroupIndex)
-			rowGroupWriter := fileWriter.AppendRowGroup()
+			fileWriter.NewRowGroup()
 			for fieldNum := 0; fieldNum < numFields; fieldNum += 1 {
 				arr, readErr := rowGroupReader.Column(fieldNum).Read(ctx)
 				if readErr != nil {
@@ -175,11 +181,7 @@ func TransformByColumn(config *TransformConfig) error {
 					}
 					arr = transformed
 				}
-				colWriter, colWriterErr := pqarrow.NewArrowColumnWriter(arr, 0, int64(arr.Len()), outputManifest, rowGroupWriter, fieldNum)
-				if colWriterErr != nil {
-					return colWriterErr
-				}
-				if err := colWriter.Write(ctx); err != nil {
+				if err := fileWriter.WriteColumnChunked(arr, 0, int64(arr.Len())); err != nil {
 					return err
 				}
 			}


Test patch:
diff --git a/internal/pqutil/transform_test.go b/internal/pqutil/transform_test.go
index 3559906..6934ea5 100644
--- a/internal/pqutil/transform_test.go
+++ b/internal/pqutil/transform_test.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/apache/arrow/go/v14/arrow"
@@ -14,6 +15,7 @@ import (
 	"github.com/apache/arrow/go/v14/parquet"
 	"github.com/apache/arrow/go/v14/parquet/compress"
 	"github.com/apache/arrow/go/v14/parquet/file"
+	"github.com/apache/arrow/go/v14/parquet/pqarrow"
 	"github.com/apache/arrow/go/v14/parquet/schema"
 	"github.com/planetlabs/gpq/internal/pqutil"
 	"github.com/planetlabs/gpq/internal/test"
@@ -123,6 +125,70 @@ func TestTransformByColumn(t *testing.T) {
 	}
 }
 
+func makeOvertureData(t *testing.T) (string, []byte) {
+	schema := arrow.NewSchema([]arrow.Field{
+		{Name: "sources", Nullable: true, Type: arrow.ListOf(arrow.StructOf(
+			arrow.Field{Name: "property", Nullable: true, Type: arrow.BinaryTypes.String},
+			arrow.Field{Name: "dataset", Nullable: true, Type: arrow.BinaryTypes.String},
+			arrow.Field{Name: "recordId", Nullable: true, Type: arrow.BinaryTypes.String},
+			arrow.Field{Name: "confidence", Nullable: true, Type: arrow.PrimitiveTypes.Float64},
+		))},
+		{Name: "bbox", Nullable: false, Type: arrow.StructOf(
+			arrow.Field{Name: "minx", Nullable: true, Type: arrow.PrimitiveTypes.Float64},
+			arrow.Field{Name: "maxx", Nullable: true, Type: arrow.PrimitiveTypes.Float64},
+			arrow.Field{Name: "miny", Nullable: true, Type: arrow.PrimitiveTypes.Float64},
+			arrow.Field{Name: "maxy", Nullable: true, Type: arrow.PrimitiveTypes.Float64},
+		)},
+	}, nil)
+
+	expected := `[
+		{
+			"sources": [
+				{
+					"property": "",
+					"recordId": "record-1",
+					"dataset": "test",
+					"confidence": null
+				}
+			],
+			"bbox": {
+				"minx": -180,
+				"maxx": -180,
+				"miny": -90,
+				"maxy": -90
+			}
+		}
+	]`
+	record, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema, strings.NewReader(expected))
+	require.NoError(t, err)
+
+	output := &bytes.Buffer{}
+	writer, err := pqarrow.NewFileWriter(schema, output, nil, pqarrow.DefaultWriterProps())
+	require.NoError(t, err)
+
+	require.NoError(t, writer.Write(record))
+	require.NoError(t, writer.Close())
+
+	return expected, output.Bytes()
+}
+
+func TestTransformOverture(t *testing.T) {
+	// minimal reproduction of https://github.com/planetlabs/gpq/issues/102
+	expected, parquetData := makeOvertureData(t)
+
+	input := bytes.NewReader(parquetData)
+	output := &bytes.Buffer{}
+	config := &pqutil.TransformConfig{
+		Reader: input,
+		Writer: output,
+	}
+
+	require.NoError(t, pqutil.TransformByColumn(config))
+
+	outputAsJSON := test.ParquetToJSON(t, bytes.NewReader(output.Bytes()))
+	assert.JSONEq(t, expected, outputAsJSON)
+}
+
 func TestTransformByRowGroupLength(t *testing.T) {
 	numRows := 100
 	rows := make([]map[string]any, numRows)
diff --git a/internal/validator/validator_test.go b/internal/validator/validator_test.go
index adffb37..f1b155f 100644
--- a/internal/validator/validator_test.go
+++ b/internal/validator/validator_test.go
@@ -28,6 +28,7 @@ import (
 
 	"github.com/apache/arrow/go/v14/parquet"
 	"github.com/apache/arrow/go/v14/parquet/file"
+	"github.com/apache/arrow/go/v14/parquet/pqarrow"
 	"github.com/paulmach/orb"
 	"github.com/paulmach/orb/encoding/wkb"
 	"github.com/planetlabs/gpq/internal/geojson"
@@ -87,7 +88,7 @@ func (s *Suite) copyWithMetadata(input parquet.ReaderAtSeeker, output io.Writer,
 	config := &pqutil.TransformConfig{
 		Reader: input,
 		Writer: output,
-		BeforeClose: func(fileReader *file.Reader, fileWriter *file.Writer) error {
+		BeforeClose: func(fileReader *file.Reader, fileWriter *pqarrow.FileWriter) error {
 			return fileWriter.AppendKeyValueMetadata(geoparquet.MetadataKey, metadata)
 		},
 	}