fix: Default max_row_group_length to a row count so parquet files split into row groups

parquet/spec.go

@@ -9,15 +9,16 @@ import (
 	"github.com/invopop/jsonschema"
 )
 
-const defaultMaxRowGroupLength = 128 * 1024 * 1024
+const defaultMaxRowGroupLength = 1024 * 1024
 
 var allowedVersions = []string{"v1.0", "v2.4", "v2.6", "v2Latest"}
 var allowedRootRepetitions = []string{"undefined", "required", "optional", "repeated"}
 
 // nolint:revive
 type ParquetSpec struct {
-	Version           string `json:"version,omitempty"`
-	RootRepetition    string `json:"root_repetition,omitempty"`
+	Version        string `json:"version,omitempty"`
+	RootRepetition string `json:"root_repetition,omitempty"`
+	// MaxRowGroupLength is the maximum number of rows in a single Parquet row group.
 	MaxRowGroupLength *int64 `json:"max_row_group_length,omitempty"`
 }
 
@@ -82,7 +83,7 @@ func (ParquetSpec) JSONSchema() *jsonschema.Schema {
 
 	properties.Set("max_row_group_length", &jsonschema.Schema{
 		Type:        "integer",
-		Description: "Max row group length",
+		Description: "Maximum number of rows per Parquet row group.",
 		Default:     defaultMaxRowGroupLength,
 		Minimum:     "0",
 	})

parquet/write_read_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/parquet/file"
 	"github.com/cloudquery/filetypes/v4/types"
 	"github.com/cloudquery/plugin-sdk/v4/plugin"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
@@ -149,3 +150,33 @@ func BenchmarkWrite(b *testing.B) {
 		b.Fatal(err)
 	}
 }
+
+func TestWriteMultipleRowGroups(t *testing.T) {
+	const rows = 20
+	const maxRowGroupLength = 5
+
+	table := schema.TestTable("test", schema.TestSourceOptions{})
+	tg := schema.NewTestDataGenerator(0)
+	record := tg.Generate(table, schema.GenTestDataOptions{
+		SourceName: "test-source",
+		SyncTime:   time.Now().UTC().Round(time.Second),
+		MaxRows:    rows,
+	})
+
+	maxRowGroupLengthValue := int64(maxRowGroupLength)
+	cl, err := NewClient(WithSpec(ParquetSpec{MaxRowGroupLength: &maxRowGroupLengthValue}))
+	require.NoError(t, err)
+
+	var b bytes.Buffer
+	writer := bufio.NewWriter(&b)
+	require.NoError(t, types.WriteAll(cl, writer, table, []arrow.RecordBatch{record}))
+	require.NoError(t, writer.Flush())
+
+	pf, err := file.NewParquetReader(bytes.NewReader(b.Bytes()))
+	require.NoError(t, err)
+	defer pf.Close()
+
+	require.Equal(t, rows/maxRowGroupLength, pf.NumRowGroups())
+	require.Greater(t, pf.NumRowGroups(), 1)
+	require.EqualValues(t, rows, pf.NumRows())
+}

schema.json

@@ -155,8 +155,8 @@
         "max_row_group_length": {
           "type": "integer",
           "minimum": 0,
-          "description": "Max row group length",
-          "default": 134217728
+          "description": "Maximum number of rows per Parquet row group.",
+          "default": 1048576
         }
       },
       "additionalProperties": false,