Reintroduce type renaming with more resilient handling for unexpected protobuf names

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/expressions/FieldKeyExpression.java

@@ -33,6 +33,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.KeyExpressionVisitor;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.ExplodeExpression;
+import com.apple.foundationdb.record.util.ProtoUtils;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.Message;
@@ -57,6 +58,13 @@
 public class FieldKeyExpression extends BaseKeyExpression implements AtomKeyExpression, KeyExpressionWithoutChildren {
     private static final ObjectPlanHash BASE_HASH = new ObjectPlanHash("Field-Key-Expression");
 
+    /**
+     * The internal field name used in the underlying protobuf message. This name may differ from the user-visible
+     * identifier to ensure compliance with protobuf field naming conventions. When working with query planning or
+     * user-facing operations, use {@link ProtoUtils#toUserIdentifier(String)} to convert this to the user-visible
+     * form. However, when constructing physical operators that directly interact with stored protobuf messages,
+     * this internal name should be used as-is.
+     */
     @Nonnull
     private final String fieldName;
     @Nonnull
@@ -216,7 +224,7 @@ public <S extends KeyExpressionVisitor.State, R> R expand(@Nonnull final KeyExpr
     public Quantifier.ForEach explodeField(@Nonnull Quantifier.ForEach baseQuantifier, @Nonnull List<String> fieldNamePrefix) {
         final List<String> fieldNames = ImmutableList.<String>builder()
                 .addAll(fieldNamePrefix)
-                .add(fieldName)
+                .add(ProtoUtils.toUserIdentifier(fieldName))
                 .build();
         switch (fanType) {
             case FanOut:

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

@@ -40,6 +40,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.values.EmptyValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
+import com.apple.foundationdb.record.util.ProtoUtils;
 import com.google.common.base.Verify;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -119,7 +120,7 @@ public GraphExpansion visitExpression(@Nonnull final EmptyKeyExpression emptyKey
     @Nonnull
     @Override
     public GraphExpansion visitExpression(@Nonnull FieldKeyExpression fieldKeyExpression) {
-        final String fieldName = fieldKeyExpression.getFieldName();
+        final String fieldName = ProtoUtils.toUserIdentifier(fieldKeyExpression.getFieldName());
         final KeyExpression.FanType fanType = fieldKeyExpression.getFanType();
         final VisitorState state = getCurrentState();
         final List<String> fieldNamePrefix = state.getFieldNamePrefix();
@@ -241,7 +242,7 @@ public GraphExpansion visitExpression(@Nonnull final NestingKeyExpression nestin
             case None:
                 List<String> newPrefix = ImmutableList.<String>builder()
                         .addAll(fieldNamePrefix)
-                        .add(parent.getFieldName())
+                        .add(ProtoUtils.toUserIdentifier((parent.getFieldName())))
                         .build();
                 if (NullableArrayTypeUtils.isArrayWrapper(nestingKeyExpression)) {
                     final RecordKeyExpressionProto.KeyExpression childProto = nestingKeyExpression.getChild().toKeyExpression();

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

@@ -35,6 +35,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.QuantifiedObjectValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
+import com.apple.foundationdb.record.util.ProtoUtils;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 
@@ -124,7 +125,7 @@ public Value visitExpression(@Nonnull FieldKeyExpression fieldKeyExpression) {
         }
 
         final ScalarVisitorState state = getCurrentState();
-        final String fieldName = fieldKeyExpression.getFieldName();
+        final String fieldName = ProtoUtils.toUserIdentifier(fieldKeyExpression.getFieldName());
         final List<String> fieldNamePrefix = state.getFieldNamePrefix();
         final List<String> fieldNames = ImmutableList.<String>builder()
                 .addAll(fieldNamePrefix)
@@ -169,9 +170,10 @@ public Value visitExpression(@Nonnull final NestingKeyExpression nestingKeyExpre
         final ScalarVisitorState state = getCurrentState();
         final List<String> fieldNamePrefix = state.getFieldNamePrefix();
         final KeyExpression child = nestingKeyExpression.getChild();
+        final String parentFieldName = ProtoUtils.toUserIdentifier(parent.getFieldName());
         final List<String> newPrefix = ImmutableList.<String>builder()
                 .addAll(fieldNamePrefix)
-                .add(parent.getFieldName())
+                .add(parentFieldName)
                 .build();
         // TODO resolve type
         return pop(child.expand(push(state.withFieldNamePrefix(newPrefix))));

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

@@ -275,7 +275,7 @@ private static List<FieldAccessTrieNodeBuilder> computeFieldAccessForDerivation(
                     Verify.verify(type.isRecord());
                     final var field = ((Type.Record)type).getField(fieldAccessor.getOrdinal());
                     currentTrieBuilder =
-                            currentTrieBuilder.compute(FieldValue.ResolvedAccessor.of(field.getFieldName(), fieldAccessor.getOrdinal(), fieldAccessor.getType()),
+                            currentTrieBuilder.compute(FieldValue.ResolvedAccessor.of(field, fieldAccessor.getOrdinal()),
                                     (resolvedAccessor, oldTrieBuilder) -> {
                                         if (oldTrieBuilder == null) {
                                             return new FieldAccessTrieNodeBuilder(null, null, field.getFieldType());