feat: avoid emit in remap when unnecessary (#84)

This PR optimizes the generated output by not providing an output
mapping when one is not needed.

It also addresses a behavior where `OutputMapping()` would return an
empty list instead of nil when there was no mapping.

plan/common.go

@@ -48,6 +48,9 @@ func (rc *RelCommon) remap(initial types.RecordType) types.RecordType {
 }
 
 func (rc *RelCommon) OutputMapping() []int32 {
+	if rc.mapping == nil {
+		return nil
+	}
 	// Make a copy of the output mapping to prevent accidental modification.
 	mapCopy := make([]int32, len(rc.mapping))
 	copy(mapCopy, rc.mapping)

plan/plan_builder_test.go

@@ -135,6 +135,18 @@ func TestMappingOfMapping(t *testing.T) {
 	assert.Equal(t, "struct<string>", newRel2.RecordType().String())
 }
 
+func TestMappingOfMappingResultingInDirectOrder(t *testing.T) {
+	b := plan.NewBuilderDefault()
+	ns := b.NamedScan([]string{"mystring, myfloat"}, baseSchema)
+	newRel, err := ns.Remap(1, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, "struct<fp32, string>", newRel.RecordType().String())
+	newRel2, err := newRel.Remap(1, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, "struct<string, fp32>", newRel2.RecordType().String())
+	assert.Equal(t, []int32(nil), newRel2.OutputMapping())
+}
+
 func TestFailedMappingOfMapping(t *testing.T) {
 	b := plan.NewBuilderDefault()
 	ns := b.NamedScan([]string{"test"}, baseSchema)

plan/relations.go

@@ -152,6 +152,17 @@ func (b *baseReadRel) updateFilters(filters []expr.Expression) {
 	b.filter, b.bestEffortFilter = filters[0], filters[1]
 }
 
+// isSequentialFromZero checks if the given slice of integers is a sequence starting from zero
+// where each element in the slice is equal to its index.
+func isSequentialFromZero(order []int32) bool {
+	for x, i := range order {
+		if i != int32(x) {
+			return false
+		}
+	}
+	return true
+}
+
 // RemapHelper implements the core functionality of RemapHelper for relations.
 func RemapHelper(r Rel, mapping []int32) (Rel, error) {
 	newRel, err := r.Copy(r.GetInputs()...)
@@ -175,7 +186,11 @@ func RemapHelper(r Rel, mapping []int32) (Rel, error) {
 			newMapping = append(newMapping, idx)
 		}
 	}
-	newRel.setMapping(newMapping)
+	if len(newMapping) == int(r.directOutputSchema().FieldCount()) && isSequentialFromZero(newMapping) {
+		newRel.setMapping(nil)
+	} else {
+		newRel.setMapping(newMapping)
+	}
 	return newRel, nil
 }