Support SQL views (#3680)

This introduces support for SQL views. Views are virtual tables defined
by SQL queries that are compiled lazily when referenced, providing a way
to encapsulate complex queries and present simplified interfaces to
users.

**Implementation Overview**

Views are stored as raw SQL text in the catalog and compiled on-demand
during query execution. The implementation follows a layered
architecture:

At the core layer, we introduce `RawView` which represents the persisted
view definition. Views are stored in the metadata alongside user-defined
functions and are serialized/deserialized using the existing protobuf
infrastructure. The `RecordMetaData` and `RecordMetaDataBuilder` classes
now maintain a map of views that get persisted with the schema.

At the relational API layer, `RecordLayerView` wraps the raw view
definition with compilation capabilities. Each view contains a lazy
compilation function that transforms the SQL text into a logical plan
when the view is first referenced. This compiled plan can be cached to
avoid recompilation on subsequent accesses.

The query parser has been extended to recognize view references in
`FROM` clauses. When a view is encountered, the semantic analyzer
retrieves the view definition from the catalog and triggers compilation
by invoking the view's compilation function. The resulting logical
operator is then seamlessly integrated into the query plan, allowing
views to be used just like regular tables.

**Features**

- View Definition Storage: Views are defined using `CREATE VIEW` syntax
and stored as SQL text in the schema template
- Lazy Compilation: View definitions are compiled into logical plans
only when referenced in queries
- Nested Views: Views can reference other views, enabling layered
abstractions
- Complex Queries: Views support CTEs, joins, aggregations, and other
advanced SQL features
- Schema Integration: Views are first-class metadata objects alongside
tables and functions

The implementation includes comprehensive testing through YAML
integration tests that validate various view scenarios including simple
filters, nested views, CTEs within views, and views referencing
user-defined functions.

This fixes #3670.

Author: hatyo

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/RecordMetaData.java

@@ -31,6 +31,7 @@
 import com.apple.foundationdb.record.metadata.UnnestedRecordType;
 import com.apple.foundationdb.record.metadata.expressions.KeyExpression;
 import com.apple.foundationdb.record.metadata.expressions.LiteralKeyExpression;
+import com.apple.foundationdb.record.metadata.View;
 import com.apple.foundationdb.record.query.plan.cascades.UserDefinedFunction;
 import com.apple.foundationdb.record.query.plan.cascades.typing.Type;
 import com.apple.foundationdb.record.query.plan.serialization.DefaultPlanSerializationRegistry;
@@ -87,6 +88,8 @@ public class RecordMetaData implements RecordMetaDataProvider {
     @Nonnull
     private final Map<String, UserDefinedFunction> userDefinedFunctionMap;
     @Nonnull
+    private final Map<String, View> viewMap;
+    @Nonnull
     private final Map<String, Index> indexes;
     @Nonnull
     private final Map<String, Index> universalIndexes;
@@ -117,6 +120,7 @@ protected RecordMetaData(@Nonnull RecordMetaData orig) {
                 Collections.unmodifiableMap(orig.universalIndexes),
                 Collections.unmodifiableList(orig.formerIndexes),
                 Collections.unmodifiableMap(orig.userDefinedFunctionMap),
+                Collections.unmodifiableMap(orig.viewMap),
                 orig.splitLongRecords,
                 orig.storeRecordVersions,
                 orig.version,
@@ -137,6 +141,7 @@ protected RecordMetaData(@Nonnull Descriptors.FileDescriptor recordsDescriptor,
                              @Nonnull Map<String, Index> universalIndexes,
                              @Nonnull List<FormerIndex> formerIndexes,
                              @Nonnull Map<String, UserDefinedFunction> userDefinedFunctionMap,
+                             @Nonnull Map<String, View> viewMap,
                              boolean splitLongRecords,
                              boolean storeRecordVersions,
                              int version,
@@ -154,6 +159,7 @@ protected RecordMetaData(@Nonnull Descriptors.FileDescriptor recordsDescriptor,
         this.universalIndexes = universalIndexes;
         this.formerIndexes = formerIndexes;
         this.userDefinedFunctionMap = userDefinedFunctionMap;
+        this.viewMap = viewMap;
         this.splitLongRecords = splitLongRecords;
         this.storeRecordVersions = storeRecordVersions;
         this.version = version;
@@ -702,6 +708,7 @@ public RecordMetaDataProto.MetaData toProto(@Nullable Descriptors.FileDescriptor
         PlanSerializationContext serializationContext = new PlanSerializationContext(DefaultPlanSerializationRegistry.INSTANCE,
                 PlanHashable.CURRENT_FOR_CONTINUATION);
         builder.addAllUserDefinedFunctions(userDefinedFunctionMap.values().stream().map(func -> func.toProto(serializationContext)).collect(Collectors.toList()));
+        builder.addAllViews(viewMap.values().stream().map(View::toProto).collect(Collectors.toList()));
         builder.setSplitLongRecords(splitLongRecords);
         builder.setStoreRecordVersions(storeRecordVersions);
         builder.setVersion(version);
@@ -726,6 +733,11 @@ public Map<String, UserDefinedFunction> getUserDefinedFunctionMap() {
         return userDefinedFunctionMap;
     }
 
+    @Nonnull
+    public Map<String, View> getViewMap() {
+        return viewMap;
+    }
+
     @Nonnull
     public static Map<String, Descriptors.FieldDescriptor> getFieldDescriptorMapFromTypes(@Nonnull final Collection<RecordType> recordTypes) {
         if (recordTypes.size() == 1) {

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/RecordMetaDataBuilder.java

@@ -42,6 +42,7 @@
 import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerRegistry;
 import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerFactoryRegistryImpl;
 import com.apple.foundationdb.record.provider.foundationdb.MetaDataProtoEditor;
+import com.apple.foundationdb.record.metadata.View;
 import com.apple.foundationdb.record.query.plan.cascades.UserDefinedFunction;
 import com.apple.foundationdb.record.query.plan.serialization.DefaultPlanSerializationRegistry;
 import com.apple.foundationdb.record.query.plan.serialization.PlanSerialization;
@@ -115,6 +116,8 @@ public class RecordMetaDataBuilder implements RecordMetaDataProvider {
     @Nonnull
     private final Map<String, UserDefinedFunction> userDefinedFunctionMap;
     @Nonnull
+    private final Map<String, View> viewMap;
+    @Nonnull
     private final Map<String, Index> indexes;
     @Nonnull
     private final Map<String, Index> universalIndexes;
@@ -150,6 +153,7 @@ public class RecordMetaDataBuilder implements RecordMetaDataProvider {
         evolutionValidator = MetaDataEvolutionValidator.getDefaultInstance();
         syntheticRecordTypes = new HashMap<>();
         userDefinedFunctionMap = new HashMap<>();
+        viewMap = new HashMap<>();
     }
 
     private void processSchemaOptions(boolean processExtensionOptions) {
@@ -234,6 +238,10 @@ private void loadProtoExceptRecords(@Nonnull RecordMetaDataProto.MetaData metaDa
                             PlanHashable.CURRENT_FOR_CONTINUATION), function);
             userDefinedFunctionMap.put(func.getFunctionName(), func);
         }
+        for (final RecordMetaDataProto.PView viewProto: metaDataProto.getViewsList()) {
+            final View view = View.fromProto(viewProto);
+            viewMap.put(view.getName(), view);
+        }
         if (metaDataProto.hasSplitLongRecords()) {
             splitLongRecords = metaDataProto.getSplitLongRecords();
         }
@@ -1207,6 +1215,10 @@ public void addUserDefinedFunctions(@Nonnull Iterable<? extends UserDefinedFunct
         functions.forEach(this::addUserDefinedFunction);
     }
 
+    public void addView(@Nonnull View view) {
+        viewMap.put(view.getName(), view);
+    }
+
     public boolean isSplitLongRecords() {
         return splitLongRecords;
     }
@@ -1448,7 +1460,7 @@ public RecordMetaData build(boolean validate) {
         Map<Object, SyntheticRecordType<?>> recordTypeKeyToSyntheticRecordTypeMap = Maps.newHashMapWithExpectedSize(syntheticRecordTypes.size());
         RecordMetaData metaData = new RecordMetaData(recordsDescriptor, getUnionDescriptor(), unionFields,
                 builtRecordTypes, builtSyntheticRecordTypes, recordTypeKeyToSyntheticRecordTypeMap,
-                indexes, universalIndexes, formerIndexes, userDefinedFunctionMap,
+                indexes, universalIndexes, formerIndexes, userDefinedFunctionMap, viewMap,
                 splitLongRecords, storeRecordVersions, version, subspaceKeyCounter, usesSubspaceKeyCounter, recordCountKey, localFileDescriptor != null);
         for (RecordTypeBuilder recordTypeBuilder : recordTypes.values()) {
             KeyExpression primaryKey = recordTypeBuilder.getPrimaryKey();

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

@@ -0,0 +1,93 @@
+/*
+ * View.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.apple.foundationdb.record.metadata;
+
+import com.apple.foundationdb.record.RecordMetaDataProto;
+
+import javax.annotation.Nonnull;
+import java.util.Objects;
+
+/**
+ * Represents a SQL view definition stored in the catalog.
+ * Views are stored as raw SQL text and compiled lazily when referenced in queries.
+ * This class is used for serialization/deserialization of view definitions.
+ */
+public class View {
+
+    @Nonnull
+    private final String name;
+
+    @Nonnull
+    private final String definition;
+
+    public View(@Nonnull final String name, @Nonnull final String definition) {
+        this.name = name;
+        this.definition = definition;
+    }
+
+    @Nonnull
+    public String getName() {
+        return name;
+    }
+
+    @Nonnull
+    public String getDefinition() {
+        return definition;
+    }
+
+    @Nonnull
+    public RecordMetaDataProto.PView toProto() {
+        return RecordMetaDataProto.PView.newBuilder()
+                .setName(name)
+                .setDefinition(definition)
+                .build();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final View that = (View) o;
+        return Objects.equals(name, that.name) && Objects.equals(definition, that.definition);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name, definition);
+    }
+
+    @Override
+    public String toString() {
+        return "View{" +
+                "name='" + name + '\'' +
+                ", definition='" + definition + '\'' +
+                '}';
+    }
+
+    @Nonnull
+    public static View fromProto(@Nonnull final RecordMetaDataProto.PView sqlView) {
+        return new View(sqlView.getName(), sqlView.getDefinition());
+    }
+}

fdb-record-layer-core/src/main/proto/record_metadata.proto

@@ -195,6 +195,7 @@ message MetaData {
   repeated JoinedRecordType joined_record_types = 12;
   repeated UnnestedRecordType unnested_record_types = 13;
   repeated PUserDefinedFunction user_defined_functions = 14;
+  repeated PView views = 15;
   extensions 1000 to 2000;
 }
 
@@ -205,6 +206,11 @@ message PUserDefinedFunction {
   }
 }
 
+message PView {
+  optional string name = 1;
+  optional string definition = 2;
+}
+
 message JoinedRecordType {
   optional string name = 1;
   optional com.apple.foundationdb.record.expressions.Value record_type_key = 4;

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/metadata/RecordMetaDataBuilderTest.java

@@ -850,4 +850,43 @@ void canSerializeAndDeserializeSyntheticRecordTypes() {
             assertEquals(expectedIndex.getRootExpression(), actualIndex.getRootExpression(), "Incorrect index root expression");
         }
     }
+
+    @Test
+    void recordMetadataCopyConstructorWorks() {
+        // Create a sample RecordMetaData object.
+        RecordMetaDataBuilder builder = RecordMetaData.newBuilder().setRecords(TestRecords1Proto.getDescriptor());
+        builder.setSplitLongRecords(true);
+        builder.setStoreRecordVersions(true);
+        builder.setVersion(42);
+
+        final RecordMetaData original = builder.build();
+
+        // Create a subclass to access the protected copy constructor
+        final RecordMetaData copy = new RecordMetaData(original) {
+        };
+
+        // Verify that all properties are correctly copied
+        assertEquals(original.getVersion(), copy.getVersion(), "Version should match");
+        assertEquals(original.isSplitLongRecords(), copy.isSplitLongRecords(), "splitLongRecords should match");
+        assertEquals(original.isStoreRecordVersions(), copy.isStoreRecordVersions(), "storeRecordVersions should match");
+        assertEquals(original.usesSubspaceKeyCounter(), copy.usesSubspaceKeyCounter(), "usesSubspaceKeyCounter should match");
+        assertEquals(original.getSubspaceKeyCounter(), copy.getSubspaceKeyCounter(), "subspaceKeyCounter should match");
+
+        // Verify record types
+        assertEquals(original.getRecordTypes().size(), copy.getRecordTypes().size(), "Record type count should match");
+        assertIterableEquals(original.getRecordTypes().keySet(), copy.getRecordTypes().keySet(), "Record type names should match");
+
+        // Verify indexes
+        assertEquals(original.getAllIndexes().size(), copy.getAllIndexes().size(), "Index count should match");
+        for (Index originalIndex : original.getAllIndexes()) {
+            Index copiedIndex = copy.getIndex(originalIndex.getName());
+            assertNotNull(copiedIndex, "Index should exist in copy: " + originalIndex.getName());
+            assertEquals(originalIndex.getName(), copiedIndex.getName(), "Index name should match");
+            assertEquals(originalIndex.getRootExpression(), copiedIndex.getRootExpression(), "Index root expression should match");
+        }
+
+        // Verify descriptors
+        assertEquals(original.getRecordsDescriptor(), copy.getRecordsDescriptor(), "Records descriptor should match");
+        assertEquals(original.getUnionDescriptor(), copy.getUnionDescriptor(), "Union descriptor should match");
+    }
 }

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/metadata/ViewTest.java

@@ -0,0 +1,121 @@
+/*
+ * ViewTest.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.apple.foundationdb.record.metadata;
+
+import com.apple.foundationdb.record.RecordMetaDataProto;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test cases for {@link View}.
+ */
+public class ViewTest {
+
+    @Test
+    public void testConstructorAndGetters() {
+        final String viewName = "test_view";
+        final String definition = "SELECT * FROM table1";
+
+        final View view = new View(viewName, definition);
+
+        assertEquals(viewName, view.getName());
+        assertEquals(definition, view.getDefinition());
+    }
+
+    @Test
+    public void testEqualsAndHashCode() {
+        final View view1 = new View("view1", "SELECT * FROM t1");
+        final View view2 = new View("view1", "SELECT * FROM t1");
+        final View view3 = new View("view2", "SELECT * FROM t1");
+        final View view4 = new View("view1", "SELECT * FROM t2");
+
+        // Test equality
+        assertEquals(view1, view2);
+        assertEquals(view1.hashCode(), view2.hashCode());
+
+        // Test inequality - different names
+        assertNotEquals(view1, view3);
+
+        // Test inequality - different definitions
+        assertNotEquals(view1, view4);
+
+        // Test self equality
+        assertEquals(view1, view1);
+
+        // Test null inequality
+        assertNotEquals(view1, null);
+    }
+
+    @Test
+    public void testToString() {
+        final View view = new View("my_view", "SELECT id, name FROM users");
+        final String result = view.toString();
+
+        assertTrue(result.contains("my_view"));
+        assertTrue(result.contains("SELECT id, name FROM users"));
+    }
+
+    @Test
+    public void testProtoSerialization() {
+        final String viewName = "serialization_test";
+        final String definition = "SELECT a, b FROM table WHERE c > 10";
+        final View originalView = new View(viewName, definition);
+
+        final RecordMetaDataProto.PView proto = originalView.toProto();
+
+        assertEquals(viewName, proto.getName());
+        assertEquals(definition, proto.getDefinition());
+    }
+
+    @Test
+    public void testProtoDeserialization() {
+        final String viewName = "deserialization_test";
+        final String definition = "SELECT x, y, z FROM test_table";
+
+        final RecordMetaDataProto.PView protoView = RecordMetaDataProto.PView.newBuilder()
+                .setName(viewName)
+                .setDefinition(definition)
+                .build();
+
+        final View view = View.fromProto(protoView);
+
+        assertEquals(viewName, view.getName());
+        assertEquals(definition, view.getDefinition());
+    }
+
+    @Test
+    public void testRoundTripSerialization() {
+        final View originalView = new View("round_trip_view", "SELECT * FROM employees WHERE salary > 50000");
+
+        // Serialize to proto
+        final RecordMetaDataProto.PView proto = originalView.toProto();
+
+        // Deserialize from proto
+        final View deserializedView = View.fromProto(proto);
+
+        // Verify equality
+        assertEquals(originalView.getName(), deserializedView.getName());
+        assertEquals(originalView.getDefinition(), deserializedView.getDefinition());
+    }
+}