fix(arrow/array): preserve rows for empty-schema JSON input

arrow/array/json_reader.go

@@ -184,19 +184,29 @@ func (r *JSONReader) readNext() bool {
 }
 
 func (r *JSONReader) nextall() bool {
+	n := 0
 	for r.readNext() {
+		n++
 	}
 
-	r.cur = r.bldr.NewRecordBatch()
-	return r.cur.NumRows() > 0
+	if r.schema.NumFields() == 0 {
+		r.cur = NewRecordBatch(r.schema, nil, int64(n))
+	} else {
+		r.cur = r.bldr.NewRecordBatch()
+	}
+	return n > 0
 }
 
 func (r *JSONReader) next1() bool {
 	if !r.readNext() {
 		return false
 	}
 
-	r.cur = r.bldr.NewRecordBatch()
+	if r.schema.NumFields() == 0 {
+		r.cur = NewRecordBatch(r.schema, nil, 1)
+	} else {
+		r.cur = r.bldr.NewRecordBatch()
+	}
 	return true
 }
 
@@ -210,7 +220,11 @@ func (r *JSONReader) nextn() bool {
 	}
 
 	if n > 0 {
-		r.cur = r.bldr.NewRecordBatch()
+		if r.schema.NumFields() == 0 {
+			r.cur = NewRecordBatch(r.schema, nil, int64(n))
+		} else {
+			r.cur = r.bldr.NewRecordBatch()
+		}
 	}
 	return n > 0
 }

arrow/array/json_reader_test.go

@@ -95,6 +95,34 @@ func TestJSONReaderAll(t *testing.T) {
 	assert.False(t, rdr.Next())
 }
 
+func TestJSONReaderPreservesRowsForEmptySchema(t *testing.T) {
+	schema := arrow.NewSchema(nil, nil)
+	mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+	defer mem.AssertSize(t, 0)
+
+	t.Run("one row per batch", func(t *testing.T) {
+		rdr := array.NewJSONReader(strings.NewReader("{} {}"), schema, array.WithAllocator(mem))
+		defer rdr.Release()
+
+		assert.True(t, rdr.Next())
+		assert.EqualValues(t, 1, rdr.RecordBatch().NumRows())
+		assert.True(t, rdr.Next())
+		assert.EqualValues(t, 1, rdr.RecordBatch().NumRows())
+		assert.False(t, rdr.Next())
+		assert.NoError(t, rdr.Err())
+	})
+
+	t.Run("all rows in one batch", func(t *testing.T) {
+		rdr := array.NewJSONReader(strings.NewReader("{} {}"), schema, array.WithAllocator(mem), array.WithChunk(-1))
+		defer rdr.Release()
+
+		assert.True(t, rdr.Next())
+		assert.EqualValues(t, 2, rdr.RecordBatch().NumRows())
+		assert.False(t, rdr.Next())
+		assert.NoError(t, rdr.Err())
+	})
+}
+
 func TestJSONReaderChunked(t *testing.T) {
 	schema := arrow.NewSchema([]arrow.Field{
 		{Name: "region", Type: arrow.BinaryTypes.String, Nullable: true},