Keep the field indexes two record types agree on when taking their maximum

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/typing/Type.java

@@ -643,8 +643,10 @@
                 if (t1Fields.size() != t2Fields.size()) {
                     return null;
                 }
 
-                final var resultFieldsBuilder = ImmutableList.<Type.Record.Field>builder();
+                final var resultFieldTypes = new ArrayList<Type>(t1Fields.size());
+                final var resultFieldNameOptionals = new ArrayList<Optional<String>>(t1Fields.size());
+                final var resultFieldIndexOptionals = new ArrayList<Optional<Integer>>(t1Fields.size());
                 for (int i = 0; i < t1Fields.size(); i++) {
                     final var t1Field = t1Fields.get(i);
                     final var t2Field = t2Fields.get(i);
@@ -653,15 +655,30 @@
                     if (resultFieldType == null) {
                         return null;
                     }
+                    resultFieldTypes.add(resultFieldType);
 
                     Optional<String> resultFieldNameOptional = Optional.empty();
                     if (t1Field.getFieldNameOptional().isEmpty()) {
                         resultFieldNameOptional = t2Field.getFieldNameOptional();
                     } else if (t2Field.getFieldNameOptional().isEmpty() || (t1Field.getFieldNameOptional().equals(t2Field.getFieldNameOptional()))) {
                         resultFieldNameOptional = t1Field.getFieldNameOptional();
                     }
+                    resultFieldNameOptionals.add(resultFieldNameOptional);
 
-                    resultFieldsBuilder.add(Record.Field.of(resultFieldType, resultFieldNameOptional));
+                    resultFieldIndexOptionals.add(
+                            t1Field.getFieldIndexOptional().equals(t2Field.getFieldIndexOptional())
+                            ? t1Field.getFieldIndexOptional()
+                            : Optional.empty());
+                }
+                // best-case effort to preserve the index of the Type.Record.Field. That is, the index is kept only if
+                // the two corresponding fields from `Type.Record`s have the same index.
+                final var keepFieldIndexes = resultFieldIndexOptionals.stream().allMatch(Optional::isPresent);
+                final var resultFieldsBuilder = ImmutableList.<Type.Record.Field>builder();
+                for (int i = 0; i < resultFieldTypes.size(); i++) {
+                    resultFieldsBuilder.add(keepFieldIndexes
+                                            ? Record.Field.of(resultFieldTypes.get(i), resultFieldNameOptionals.get(i),
+                                                    resultFieldIndexOptionals.get(i))
+                                            : Record.Field.of(resultFieldTypes.get(i), resultFieldNameOptionals.get(i)));
                 }
                 return Type.Record.fromFields(isResultNullable, resultFieldsBuilder.build());
 

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/cascades/TypeTest.java

@@ -1361,4 +1361,39 @@ void testFunctionTypeIsSingleton() {
         Assertions.assertSame(function1, function2,
                 "Type.FUNCTION should be a singleton");
     }
+
+    @Nonnull
+    private static Type.Record recordWithFieldIndexes(@Nonnull final List<Integer> fieldIndexes) {
+        final var fields = ImmutableList.<Type.Record.Field>builder();
+        for (int i = 0; i < fieldIndexes.size(); i++) {
+            fields.add(Type.Record.Field.of(Type.primitiveType(Type.TypeCode.STRING, true),
+                    Optional.of("f" + i), Optional.of(fieldIndexes.get(i))));
+        }
+        return Type.Record.fromFields(fields.build());
+    }
+
+    @Nonnull
+    private static List<Integer> fieldIndexesOf(@Nullable final Type type) {
+        return Objects.requireNonNull((Type.Record)type).getFields()
+                .stream()
+                .map(Type.Record.Field::getFieldIndex)
+                .toList();
+    }
+
+    @Test
+    void maximumTypeKeepsFieldIndexesThatBothSidesAgreeOn() {
+        final var skipping = recordWithFieldIndexes(List.of(1, 3, 4));
+        Assertions.assertEquals(List.of(1, 3, 4), fieldIndexesOf(Type.maximumType(skipping, skipping)));
+    }
+
+    @Test
+    void maximumTypeRenumbersFieldIndexesThatDisagree() {
+        final var skipping = recordWithFieldIndexes(List.of(1, 3, 4));
+        final var otherSkipping = recordWithFieldIndexes(List.of(1, 3, 5));
+
+        // Nothing can be said about the numbering of the result, so it is numbered by position, as it always was.
+        Assertions.assertEquals(List.of(1, 2, 3), fieldIndexesOf(Type.maximumType(skipping, otherSkipping)));
+        Assertions.assertEquals(List.of(1, 2, 3),
+                fieldIndexesOf(Type.maximumType(skipping, recordWithFieldIndexes(List.of(1, 2, 3)))));
+    }
 }