SERVER-63123 Add support for creating columnar indexes with a subset of fields via projection

26 files changed, 630 insertions, 84 deletions

diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 837fe9f52e3..7da78fa5703 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -764,12 +764,8 @@ Status validateColumnStoreSpec(const CollectionPtr& collection,
                 "collection"};
     }
 
-    // TODO SERVER-63123 support 'columnstoreProjection'.
-    for (auto&& notToBeSpecified : {"sparse"_sd,
-                                    "unique"_sd,
-                                    "expireAfterSeconds"_sd,
-                                    "partialFilterExpression"_sd,
-                                    "columnstoreProjection"_sd}) {
+    for (auto&& notToBeSpecified :
+         {"sparse"_sd, "unique"_sd, "expireAfterSeconds"_sd, "partialFilterExpression"_sd}) {
         if (spec.hasField(notToBeSpecified)) {
             return reportInvalidOption(notToBeSpecified, IndexNames::COLUMN);
         }
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index 199b1f5a13d..8fd78cf2084 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -40,6 +40,7 @@
 #include "mongo/base/status.h"
 #include "mongo/base/status_with.h"
 #include "mongo/db/field_ref.h"
+#include "mongo/db/index/column_key_generator.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index/wildcard_key_generator.h"
 #include "mongo/db/index_names.h"
@@ -192,10 +193,17 @@ Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion inde
                                         << "' index must be a non-zero number, not a string.");
         }
 
-        // Some special index types do not support compound indexes.
-        if (key.nFields() != 1 &&
-            (pluginName == IndexNames::WILDCARD || pluginName == IndexNames::COLUMN)) {
-            return Status(code, str::stream() << pluginName << " indexes do not allow compounding");
+        if (pluginName == IndexNames::WILDCARD || pluginName == IndexNames::COLUMN) {
+            StringData fieldName(keyElement.fieldNameStringData());
+            if (key.nFields() != 1) {
+                // Some special index types do not support compound indexes.
+                return Status(code,
+                              str::stream() << pluginName << " indexes do not allow compounding");
+            } else if ((fieldName != "$**") && !fieldName.endsWith(".$**")) {
+                // Invalid key names for wildcard or columnstore are not supported.
+                return Status(code,
+                              str::stream() << "Invalid key name for " << pluginName << " indexes");
+            }
         }
 
         // Ensure that the fields on which we are building the index are valid: a field must not
@@ -210,7 +218,7 @@ Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion inde
             return Status(code, "Index keys cannot be an empty field.");
         }
 
-        // "$**" is acceptable for a text index or wildcard index.
+        // "$**" is acceptable for a text, wildcard, or columnstore index.
         if ((keyElement.fieldNameStringData() == "$**") &&
             ((keyElement.isNumber()) || (keyElement.str() == IndexNames::TEXT) ||
              (keyElement.str() == IndexNames::COLUMN)))
@@ -237,10 +245,10 @@ Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion inde
             const bool mightBePartOfDbRef =
                 (i != 0) && (part == "$db" || part == "$id" || part == "$ref");
 
-            const bool isPartOfWildcard =
-                (i == numParts - 1) && (part == "$**") && (pluginName == IndexNames::WILDCARD);
+            const bool isWildcardOrColumn = (i == numParts - 1) && (part == "$**") &&
+                (pluginName == IndexNames::WILDCARD || pluginName == IndexNames::COLUMN);
 
-            if (!mightBePartOfDbRef && !isPartOfWildcard) {
+            if (!mightBePartOfDbRef && !isWildcardOrColumn) {
                 return Status(code,
                               "Index key contains an illegal field name: "
                               "field name starts with '$'.");
@@ -455,40 +463,51 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
             if (!statusWithMatcher.isOK()) {
                 return statusWithMatcher.getStatus();
             }
-        } else if (IndexDescriptor::kPathProjectionFieldName == indexSpecElemFieldName) {
+        } else if (IndexDescriptor::kWildcardProjectionFieldName == indexSpecElemFieldName ||
+                   IndexDescriptor::kColumnStoreProjectionFieldName == indexSpecElemFieldName) {
+            const bool isWildcard =
+                IndexDescriptor::kWildcardProjectionFieldName == indexSpecElemFieldName;
+            const auto indexName = isWildcard ? IndexNames::WILDCARD : IndexNames::COLUMN;
             const auto key = indexSpec.getObjectField(IndexDescriptor::kKeyPatternFieldName);
-            if (IndexNames::findPluginName(key) != IndexNames::WILDCARD) {
-                return {ErrorCodes::BadValue,
-                        str::stream()
-                            << "The field '" << IndexDescriptor::kPathProjectionFieldName
-                            << "' is only allowed in an '" << IndexNames::WILDCARD << "' index"};
+            if (IndexNames::findPluginName(key) != indexName) {
+                // For backwards compatibility, we will return BadValue for Wildcard indices.
+                auto code =
+                    isWildcard ? ErrorCodes::BadValue : ErrorCodes::InvalidIndexSpecificationOption;
+                return {code,
+                        str::stream() << "The field '" << indexSpecElemFieldName
+                                      << "' is only allowed in '" << indexName << "' indexes"};
             }
             if (indexSpecElem.type() != BSONType::Object) {
                 return {ErrorCodes::TypeMismatch,
-                        str::stream() << "The field '" << IndexDescriptor::kPathProjectionFieldName
+                        str::stream() << "The field '" << indexSpecElemFieldName
                                       << "' must be a non-empty object, but got "
                                       << typeName(indexSpecElem.type())};
             }
             if (!key.hasField("$**")) {
                 return {ErrorCodes::FailedToParse,
                         str::stream()
-                            << "The field '" << IndexDescriptor::kPathProjectionFieldName
-                            << "' is only allowed when '" << IndexDescriptor::kKeyPatternFieldName
-                            << "' is {\"$**\": ±1}"};
+                            << "The field '" << indexSpecElemFieldName << "' is only allowed when '"
+                            << IndexDescriptor::kKeyPatternFieldName
+                            << "' is {\"$**\": ±1} or {\"$**\": \"columnstore\"}"};
             }
-
             if (indexSpecElem.embeddedObject().isEmpty()) {
                 return {ErrorCodes::FailedToParse,
-                        str::stream() << "The '" << IndexDescriptor::kPathProjectionFieldName
+                        str::stream() << "The '" << indexSpecElemFieldName
                                       << "' field can't be an empty object"};
             }
             try {
-                // We use WildcardKeyGenerator::createProjectionExec to parse and validate the path
-                // projection spec.
-                WildcardKeyGenerator::createProjectionExecutor(key, indexSpecElem.embeddedObject());
+                // We use createProjectionExecutor to parse and validate the path projection spec.
+                if (isWildcard) {
+                    WildcardKeyGenerator::createProjectionExecutor(key,
+                                                                   indexSpecElem.embeddedObject());
+                } else {
+                    ColumnKeyGenerator::createProjectionExecutor(key,
+                                                                 indexSpecElem.embeddedObject());
+                }
+
             } catch (const DBException& ex) {
-                return ex.toStatus(str::stream() << "Failed to parse: "
-                                                 << IndexDescriptor::kPathProjectionFieldName);
+                return ex.toStatus(str::stream()
+                                   << "Failed to parse projection: " << indexSpecElemFieldName);
             }
         } else if (IndexDescriptor::kWeightsFieldName == indexSpecElemFieldName) {
             if (!indexSpecElem.isABSONObj() && indexSpecElem.type() != String) {
diff --git a/src/mongo/db/catalog/index_key_validate.h b/src/mongo/db/catalog/index_key_validate.h
index 8f226749c21..f9680ac035b 100644
--- a/src/mongo/db/catalog/index_key_validate.h
+++ b/src/mongo/db/catalog/index_key_validate.h
@@ -62,7 +62,8 @@ static std::set<StringData> allowedFieldNames = {
     IndexDescriptor::kLanguageOverrideFieldName,
     IndexDescriptor::kNamespaceFieldName,
     IndexDescriptor::kPartialFilterExprFieldName,
-    IndexDescriptor::kPathProjectionFieldName,
+    IndexDescriptor::kWildcardProjectionFieldName,
+    IndexDescriptor::kColumnStoreProjectionFieldName,
     IndexDescriptor::kSparseFieldName,
     IndexDescriptor::kStorageEngineFieldName,
     IndexDescriptor::kTextVersionFieldName,
diff --git a/src/mongo/db/catalog/index_key_validate_test.cpp b/src/mongo/db/catalog/index_key_validate_test.cpp
index e5b9e7b6316..9c4813a2d2e 100644
--- a/src/mongo/db/catalog/index_key_validate_test.cpp
+++ b/src/mongo/db/catalog/index_key_validate_test.cpp
@@ -291,6 +291,81 @@ TEST(IndexKeyValidateTest, KeyElementNameWildcardFailsWhenValueIsPluginNameWithV
     ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
 }
 
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreSucceedsOnSubPath) {
+    ASSERT_OK(validateKeyPattern(BSON("a.$**"
+                                      << "columnstore"),
+                                 IndexVersion::kV2));
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreSucceeds) {
+    ASSERT_OK(validateKeyPattern(BSON("$**"
+                                      << "columnstore"),
+                                 IndexVersion::kV2));
+}
+
+TEST(IndexKeyValidateTest, ColumnstoreIndexKeyElementFailsForInvalidValue) {
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateKeyPattern(BSON("$**"
+                                      << "columnStore"),
+                                 IndexVersion::kV2));
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateKeyPattern(BSON("$**"
+                                      << "notColumnstore"),
+                                 IndexVersion::kV2));
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateKeyPattern(BSON("$**"
+                                      << "column store"),
+                                 IndexVersion::kV2));
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateKeyPattern(BSON("$**"
+                                      << "cs"),
+                                 IndexVersion::kV2));
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateKeyPattern(BSON("$**"
+                                      << "string"),
+                                 IndexVersion::kV2));
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreFailsOnRepeat) {
+    auto status = validateKeyPattern(BSON("$**.$**"
+                                          << "columnstore"),
+                                     IndexVersion::kV2);
+    ASSERT_NOT_OK(status);
+    ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreFailsOnSubPathRepeat) {
+    auto status = validateKeyPattern(BSON("a.$**.$**"
+                                          << "columnstore"),
+                                     IndexVersion::kV2);
+    ASSERT_NOT_OK(status);
+    ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreFailsOnCompound) {
+    auto status = validateKeyPattern(BSON("$**"
+                                          << "columnstore"
+                                          << "a"
+                                          << "columnstore"),
+                                     IndexVersion::kV2);
+    ASSERT_NOT_OK(status);
+    ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreFailsOnIncorrectValue) {
+    auto status = validateKeyPattern(BSON("$**" << true), IndexVersion::kV2);
+    ASSERT_NOT_OK(status);
+    ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
+}
+
+TEST(IndexKeyValidateTest, KeyElementNameColumnstoreFailsWhenValueIsPluginNameWithInvalidKeyName) {
+    auto status = validateKeyPattern(BSON("a"
+                                          << "columnstore"),
+                                     IndexVersion::kV2);
+    ASSERT_NOT_OK(status);
+    ASSERT_EQ(status, ErrorCodes::CannotCreateIndex);
+}
+
 TEST(IndexKeyValidateTest, CompoundHashedIndex) {
     // Validation succeeds with hashed prefix in the index.
     ASSERT_OK(index_key_validate::validateIndexSpec(
diff --git a/src/mongo/db/catalog/index_signature_test.cpp b/src/mongo/db/catalog/index_signature_test.cpp
index 6595cfb29bf..b352bb205ca 100644
--- a/src/mongo/db/catalog/index_signature_test.cpp
+++ b/src/mongo/db/catalog/index_signature_test.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/query/collection_query_info.h"
+#include "mongo/idl/server_parameter_test_util.h"
 
 namespace mongo {
 namespace {
@@ -443,5 +444,144 @@ TEST_F(IndexSignatureTest,
         ErrorCodes::IndexOptionsConflict);
 }
 
+TEST_F(IndexSignatureTest, NormalizeOnlyColumnstoreAllKeyPattern) {
+    RAIIServerParameterControllerForTest controller("featureFlagColumnstoreIndexes", true);
+
+    auto csAInclusionSpec = fromjson("{v: 2, key: {'a.$**': 'columnstore'}}");
+
+    // Verifies that the path projection is not normalized.
+    auto csAInclusionDesc = makeIndexDescriptor(csAInclusionSpec);
+    auto csAInclusionSpecAfterNormalization = normalizeIndexSpec(csAInclusionSpec);
+    ASSERT_TRUE(SimpleBSONObjComparator::kInstance.evaluate(csAInclusionSpec ==
+                                                            csAInclusionSpecAfterNormalization));
+    // Verifies that the path projection is not normalized for the created index catalog entry.
+    auto* csAInclusionIndex = unittest::assertGet(
+        createIndex(csAInclusionSpec.addFields(fromjson("{name: 'cs_a_all'}"))));
+    ASSERT_TRUE(csAInclusionIndex->descriptor()->normalizedPathProjection().isEmpty());
+}
+
+TEST_F(IndexSignatureTest,
+       CanCreateMultipleIndexesOnSameKeyPatternWithDifferentColumnstoreProjections) {
+    RAIIServerParameterControllerForTest controller("featureFlagColumnstoreIndexes", true);
+
+    // Creates a base columnstore index to verify 'columnstoreProjection' option is part of index
+    // signature.
+    auto* columnstoreIndex = unittest::assertGet(
+        createIndex(fromjson("{v: 2, name: 'cs_all', key: {'$**': 'columnstore'}}")));
+
+    // Verifies that another columnstore index with empty columnstoreProjection compares identical
+    // to 'columnstoreIndex' after normalizing the index spec.
+    auto anotherCsAllSpec = normalizeIndexSpec(fromjson("{v: 2, key: {'$**': 'columnstore'}}"));
+    auto anotherCsAllProjDesc = makeIndexDescriptor(anotherCsAllSpec);
+
+    ASSERT(anotherCsAllProjDesc->compareIndexOptions(opCtx(), coll()->ns(), columnstoreIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+
+    ASSERT_EQ(createIndex(anotherCsAllProjDesc->infoObj().addFields(fromjson("{name: 'cs_all'}"))),
+              ErrorCodes::IndexAlreadyExists);
+    ASSERT_EQ(
+        createIndex(anotherCsAllProjDesc->infoObj().addFields(fromjson("{name: 'cs_all_1'}"))),
+        ErrorCodes::IndexOptionsConflict);
+
+    // Verifies that an index with non-empty value for 'columnstoreProjection' option compares
+    // different from the base columnstore index and thus can be created.
+    auto csProjADesc =
+        makeIndexDescriptor(normalizeIndexSpec(columnstoreIndex->descriptor()->infoObj().addFields(
+            fromjson("{columnstoreProjection: {a: 1}}"))));
+    ASSERT(csProjADesc->compareIndexOptions(opCtx(), coll()->ns(), columnstoreIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    auto* csProjAIndex = unittest::assertGet(
+        createIndex(csProjADesc->infoObj().addFields(fromjson("{name: 'cs_a'}"))));
+
+
+    // Verifies that an index with the same value for 'columnstoreProjection' option as the
+    // csProjAIndex compares identical.
+    auto anotherCsProjADesc =
+        makeIndexDescriptor(normalizeIndexSpec(columnstoreIndex->descriptor()->infoObj().addFields(
+            fromjson("{columnstoreProjection: {a: 1}}"))));
+
+    ASSERT(anotherCsProjADesc->compareIndexOptions(opCtx(), coll()->ns(), csProjAIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+
+    // Verifies that creating an index with the same value for 'columnstoreProjection' option and
+    // the same name fails with IndexAlreadyExists error.
+    ASSERT_EQ(createIndex(anotherCsProjADesc->infoObj().addFields(fromjson("{name: 'cs_a'}"))),
+              ErrorCodes::IndexAlreadyExists);
+    // Verifies that creating an index with the same value for 'columnstoreProjection' option and a
+    // different name fails with IndexOptionsConflict error.
+    ASSERT_EQ(createIndex(anotherCsProjADesc->infoObj().addFields(fromjson("{name: 'cs_a_1'}"))),
+              ErrorCodes::IndexOptionsConflict);
+
+    // Verifies that an index with a different value for 'columnstoreProjection' option compares
+    // different from the base columnstore index or 'cs_a' and thus can be created.
+    auto csProjABDesc =
+        makeIndexDescriptor(normalizeIndexSpec(columnstoreIndex->descriptor()->infoObj().addFields(
+            fromjson("{columnstoreProjection: {a: 1, b: 1}}"))));
+    ASSERT(csProjABDesc->compareIndexOptions(opCtx(), coll()->ns(), columnstoreIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    ASSERT(csProjABDesc->compareIndexOptions(opCtx(), coll()->ns(), csProjAIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    auto* csProjABIndex = unittest::assertGet(
+        createIndex(csProjABDesc->infoObj().addFields(fromjson("{name: 'cs_a_b'}"))));
+
+    // Verifies that an index with sub fields for 'columnstoreProjection' option compares
+    // different from the base columnstore index or 'cs_a' or 'cs_a_b' and thus can be created.
+    auto csProjASubBCDesc =
+        makeIndexDescriptor(normalizeIndexSpec(columnstoreIndex->descriptor()->infoObj().addFields(
+            fromjson("{columnstoreProjection: {a: {b: 1, c: 1}}}"))));
+    ASSERT(csProjASubBCDesc->compareIndexOptions(opCtx(), coll()->ns(), columnstoreIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    ASSERT(csProjASubBCDesc->compareIndexOptions(opCtx(), coll()->ns(), csProjAIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    ASSERT(csProjASubBCDesc->compareIndexOptions(opCtx(), coll()->ns(), csProjABIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+    auto* csProjASubBCIndex = unittest::assertGet(
+        createIndex(csProjASubBCDesc->infoObj().addFields(fromjson("{name: 'cs_a_sub_b_c'}"))));
+
+    // Verifies that two indexes with the same projection in different order compares identical.
+    auto csProjASubCBDesc =
+        makeIndexDescriptor(normalizeIndexSpec(columnstoreIndex->descriptor()->infoObj().addFields(
+            fromjson("{columnstoreProjection: {a: {c: 1, b: 1}}}"))));
+    ASSERT(csProjASubCBDesc->compareIndexOptions(opCtx(), coll()->ns(), csProjASubBCIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+    // Verifies that two indexes with the same projection in different order can not be created.
+    ASSERT_EQ(
+        createIndex(csProjASubCBDesc->infoObj().addFields(fromjson("{name: 'cs_a_sub_b_c'}"))),
+        ErrorCodes::IndexAlreadyExists);
+    ASSERT_EQ(
+        createIndex(csProjASubCBDesc->infoObj().addFields(fromjson("{name: 'cs_a_sub_c_b'}"))),
+        ErrorCodes::IndexOptionsConflict);
+
+    // Verifies that an index with the same value for 'columnstoreProjection' option and a
+    // non-signature index option compares equivalent as the 'csProjAIndex'
+    auto csProjAWithNonSigDesc = makeIndexDescriptor(
+        anotherCsProjADesc->infoObj().addFields(fromjson("{storageEngine: {wiredTiger: {}}}")));
+    ASSERT(csProjAWithNonSigDesc->compareIndexOptions(opCtx(), coll()->ns(), csProjAIndex) ==
+           IndexDescriptor::Comparison::kEquivalent);
+
+    // Verifies that an index with the same value for 'columnstoreProjection' option, non-signature
+    // index option, and the same name fails with IndexOptionsConflict error.
+    ASSERT_EQ(createIndex(csProjAWithNonSigDesc->infoObj().addFields(fromjson("{name: 'cs_a'}"))),
+              ErrorCodes::IndexOptionsConflict);
+    // Verifies that an index with the same value for 'columnstoreProjection' option, non-signature
+    // index option, and a different name fails with IndexOptionsConflict error too.
+    ASSERT_EQ(
+        createIndex(csProjAWithNonSigDesc->infoObj().addFields(fromjson("{name: 'cs_a_nonsig'}"))),
+        ErrorCodes::IndexOptionsConflict);
+
+    // Creates single field columnstore index to verify 'columnstoreProjection' option is part of
+    // index signature.
+    auto* singleFieldColumnstoreIndex = unittest::assertGet(
+        createIndex(fromjson("{v: 2, name: 'single_cs', key: {'a.$**': 'columnstore'}}")));
+    // Verifies that another columnstore index with empty columnstoreProjection compares identical
+    // to 'columnstoreIndex' after normalizing the index spec.
+    auto anotherSingleFieldCsAllSpec =
+        normalizeIndexSpec(fromjson("{v: 2, key: {'a.$**': 'columnstore'}}"));
+    auto anotherSingleFieldCsAllProjDesc = makeIndexDescriptor(anotherSingleFieldCsAllSpec);
+    ASSERT(anotherSingleFieldCsAllProjDesc->compareIndexOptions(
+               opCtx(), coll()->ns(), singleFieldColumnstoreIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_spec_validate_test.cpp b/src/mongo/db/catalog/index_spec_validate_test.cpp
index 094c055f7be..07fc0f642bc 100644
--- a/src/mongo/db/catalog/index_spec_validate_test.cpp
+++ b/src/mongo/db/catalog/index_spec_validate_test.cpp
@@ -634,5 +634,144 @@ TEST(IndexSpecWildcard, FailsWhenExclusionWithSubpath) {
     ASSERT_EQ(result.getStatus().code(), ErrorCodes::FailedToParse);
 }
 
+TEST(IndexSpecColumnStore, SucceedsWithInclusion) {
+    auto result =
+        validateIndexSpec(kDefaultOpCtx,
+                          BSON("key" << BSON("$**"
+                                             << "columnstore")
+                                     << "name"
+                                     << "indexName"
+                                     << "columnstoreProjection" << BSON("a" << 1 << "b" << 1)));
+    ASSERT_OK(result.getStatus());
+}
+
+TEST(IndexSpecColumnStore, SucceedsWithExclusion) {
+    auto result =
+        validateIndexSpec(kDefaultOpCtx,
+                          BSON("key" << BSON("$**"
+                                             << "columnstore")
+                                     << "name"
+                                     << "indexName"
+                                     << "columnstoreProjection" << BSON("a" << 0 << "b" << 0)));
+    ASSERT_OK(result.getStatus());
+}
+
+TEST(IndexSpecColumnStore, SucceedsWithExclusionIncludingId) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection"
+                                               << BSON("_id" << 1 << "a" << 0 << "b" << 0)));
+    ASSERT_OK(result.getStatus());
+}
+
+TEST(IndexSpecColumnStore, SucceedsWithInclusionExcludingId) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection"
+                                               << BSON("_id" << 0 << "a" << 1 << "b" << 1)));
+    ASSERT_OK(result.getStatus());
+}
+
+TEST(IndexSpecColumnStore, FailsWithInclusionExcludingIdSubfield) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection"
+                                               << BSON("_id.field" << 0 << "a" << 1 << "b" << 1)));
+    ASSERT_EQ(result.getStatus().code(), 31253);
+}
+
+TEST(IndexSpecColumnStore, FailsWithExclusionIncludingIdSubfield) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection"
+                                               << BSON("_id.field" << 1 << "a" << 0 << "b" << 0)));
+    ASSERT_EQ(result.getStatus().code(), 31254);
+}
+
+TEST(IndexSpecColumnStore, FailsWithMixedProjection) {
+    auto result =
+        validateIndexSpec(kDefaultOpCtx,
+                          BSON("key" << BSON("$**"
+                                             << "columnstore")
+                                     << "name"
+                                     << "indexName"
+                                     << "columnstoreProjection" << BSON("a" << 1 << "b" << 0)));
+    ASSERT_EQ(result.getStatus().code(), 31254);
+}
+
+TEST(IndexSpecColumnStore, FailsWithComputedFieldsInProjection) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection"
+                                               << BSON("a" << 1 << "b"
+                                                           << "string")));
+    ASSERT_EQ(result.getStatus().code(), 51271);
+}
+
+TEST(IndexSpecColumnStore, FailsWhenProjectionPluginNotColumnStore) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("a"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection" << BSON("a" << 1)));
+    ASSERT_EQ(result.getStatus().code(), ErrorCodes::CannotCreateIndex);
+}
+
+TEST(IndexSpecColumnStore, FailsWhenProjectionIsNotAnObject) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection" << 4));
+    ASSERT_EQ(result.getStatus().code(), ErrorCodes::TypeMismatch);
+}
+
+TEST(IndexSpecColumnStore, FailsWithEmptyProjection) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection" << BSONObj()));
+    ASSERT_EQ(result.getStatus().code(), ErrorCodes::FailedToParse);
+}
+
+TEST(IndexSpecColumnStore, FailsWhenInclusionWithSubpath) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("a.$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection" << BSON("a" << 1)));
+    ASSERT_EQ(result.getStatus().code(), ErrorCodes::FailedToParse);
+}
+
+TEST(IndexSpecColumnStore, FailsWhenExclusionWithSubpath) {
+    auto result = validateIndexSpec(kDefaultOpCtx,
+                                    BSON("key" << BSON("a.$**"
+                                                       << "columnstore")
+                                               << "name"
+                                               << "indexName"
+                                               << "columnstoreProjection" << BSON("b" << 0)));
+    ASSERT_EQ(result.getStatus().code(), ErrorCodes::FailedToParse);
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 117f46a8a5a..4312a142906 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -97,7 +97,8 @@ static std::set<StringData> allowedFieldNames = {
     ListIndexesReplyItem::kUniqueFieldName,
     ListIndexesReplyItem::kVFieldName,
     ListIndexesReplyItem::kWeightsFieldName,
-    ListIndexesReplyItem::kWildcardProjectionFieldName};
+    ListIndexesReplyItem::kWildcardProjectionFieldName,
+    ListIndexesReplyItem::kColumnstoreProjectionFieldName};
 
 /**
  * Returns index specs, with resolved namespace, from the catalog for this listIndexes request.
diff --git a/src/mongo/db/create_indexes.idl b/src/mongo/db/create_indexes.idl
index f00f287af5a..179571976ab 100644
--- a/src/mongo/db/create_indexes.idl
+++ b/src/mongo/db/create_indexes.idl
@@ -164,6 +164,10 @@ structs:
                 type: object_owned
                 optional: true
                 unstable: false
+            columnstoreProjection:
+                type: object_owned
+                optional: true
+                unstable: true
             coarsestIndexedLevel:
                 type: safeInt
                 optional: true
diff --git a/src/mongo/db/exec/wildcard_projection.h b/src/mongo/db/exec/index_path_projection.h
index 38005b5cf2c..3a04ec1481a 100644
--- a/src/mongo/db/exec/wildcard_projection.h
+++ b/src/mongo/db/exec/index_path_projection.h
@@ -27,14 +27,15 @@
  *    it in the license file.
  */
 
+#pragma once
 #include "mongo/db/exec/projection_executor.h"
 #include "mongo/db/field_ref.h"
 
 namespace mongo {
 
-class WildcardProjection {
+class IndexPathProjection {
 public:
-    WildcardProjection(std::unique_ptr<projection_executor::ProjectionExecutor> projExec)
+    IndexPathProjection(std::unique_ptr<projection_executor::ProjectionExecutor> projExec)
         : _exec(std::move(projExec)), _exhaustivePaths(_exec->extractExhaustivePaths()) {
         invariant(_exec);
     }
@@ -53,4 +54,8 @@ private:
     // Store this here to avoid having to recompute it repeatedly, which is expensive.
     boost::optional<std::set<FieldRef>> _exhaustivePaths;
 };
+
+using WildcardProjection = IndexPathProjection;
+using ColumnStoreProjection = IndexPathProjection;
+
 }  // namespace mongo
diff --git a/src/mongo/db/index/column_key_generator.cpp b/src/mongo/db/index/column_key_generator.cpp
index 5002236eec3..5913a543a03 100644
--- a/src/mongo/db/index/column_key_generator.cpp
+++ b/src/mongo/db/index/column_key_generator.cpp
@@ -43,6 +43,49 @@
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kIndex
 
 
+namespace mongo {
+ColumnKeyGenerator::ColumnKeyGenerator(BSONObj keyPattern, BSONObj pathProjection)
+    : _proj(createProjectionExecutor(keyPattern, pathProjection)), _keyPattern(keyPattern){};
+
+ColumnStoreProjection ColumnKeyGenerator::createProjectionExecutor(BSONObj keyPattern,
+                                                                   BSONObj pathProjection) {
+
+    // We should never have a key pattern that contains more than a single element.
+    invariant(keyPattern.nFields() == 1);
+
+    // The keyPattern is either { "$**": "columnstore" } for all paths or { "path.$**":
+    // "columnstore" } for a single subtree. If we are indexing a single subtree, then we will
+    // project just that path.
+    auto indexRoot = keyPattern.firstElement().fieldNameStringData();
+    auto suffixPos = indexRoot.find(kSubtreeSuffix);
+    auto idPresent = indexRoot.startsWith("_id."_sd);
+
+    // If we're indexing a single subtree, we can't also specify a path projection.
+    invariant(suffixPos == std::string::npos || pathProjection.isEmpty());
+
+    // If this is a subtree projection, the projection spec is { "path.to.subtree.$**":
+    // "columnstore" }. Otherwise, we use the path projection from the original command object. For
+    // the subtree projection, we exclude the "_id" field unless the subtree is rooted off of "_id".
+    BSONObj projSpec;
+    if (suffixPos != std::string::npos) {
+        auto path = indexRoot.substr(0, suffixPos);
+        projSpec = idPresent ? BSON(path << 1) : BSON(path << 1 << "_id" << 0);
+    } else {
+        projSpec = pathProjection;
+    }
+
+    // Construct a dummy ExpressionContext for ProjectionExecutor. It's OK to set the
+    // ExpressionContext's OperationContext and CollatorInterface to 'nullptr' and the namespace
+    // string to '' here; since we ban computed fields from the projection, the ExpressionContext
+    // will never be used.
+    auto expCtx = make_intrusive<ExpressionContext>(nullptr, nullptr, NamespaceString());
+    auto policies = ProjectionPolicies::columnStoreIndexSpecProjectionPolicies();
+    auto projection = projection_ast::parseAndAnalyze(expCtx, projSpec, policies);
+    return ColumnStoreProjection{projection_executor::buildProjectionExecutor(
+        expCtx, &projection, policies, projection_executor::kDefaultBuilderParams)};
+}
+}  // namespace mongo
+
 namespace mongo::column_keygen {
 namespace {
 
diff --git a/src/mongo/db/index/column_key_generator.h b/src/mongo/db/index/column_key_generator.h
index 5fa06031bd4..b9f02eadecd 100644
--- a/src/mongo/db/index/column_key_generator.h
+++ b/src/mongo/db/index/column_key_generator.h
@@ -32,9 +32,36 @@
 #include <iosfwd>
 
 #include "mongo/db/catalog/index_catalog.h"
+#include "mongo/db/exec/index_path_projection.h"
+#include "mongo/db/exec/projection_executor.h"
+#include "mongo/db/exec/projection_executor_builder.h"
+#include "mongo/db/query/projection_parser.h"
 #include "mongo/db/storage/column_store.h"
 #include "mongo/util/functional.h"
 
+namespace mongo {
+class ColumnKeyGenerator {
+public:
+    static constexpr StringData kSubtreeSuffix = ".$**"_sd;
+
+    ColumnKeyGenerator(BSONObj keyPattern, BSONObj pathProjection);
+
+    static ColumnStoreProjection createProjectionExecutor(BSONObj keyPattern,
+                                                          BSONObj pathProjection);
+
+    /**
+     * Returns a pointer to the key generator's underlying ProjectionExecutor.
+     */
+    const ColumnStoreProjection* getColumnstoreProjection() const {
+        return &_proj;
+    }
+
+private:
+    ColumnStoreProjection _proj;
+    const BSONObj _keyPattern;
+};
+}  // namespace mongo
+
 namespace mongo::column_keygen {
 /**
  * This is a representation of the cell prior to flattening it out into a buffer which is passed to
diff --git a/src/mongo/db/index/columns_access_method.cpp b/src/mongo/db/index/columns_access_method.cpp
index b29ec20967a..b4c21443591 100644
--- a/src/mongo/db/index/columns_access_method.cpp
+++ b/src/mongo/db/index/columns_access_method.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/index/column_cell.h"
 #include "mongo/db/index/column_key_generator.h"
 #include "mongo/db/index/column_store_sorter.h"
+#include "mongo/db/index/index_descriptor.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/progress_meter.h"
 
@@ -56,7 +57,18 @@ inline void inc(int64_t* counter) {
 
 ColumnStoreAccessMethod::ColumnStoreAccessMethod(IndexCatalogEntry* ice,
                                                  std::unique_ptr<ColumnStore> store)
-    : _store(std::move(store)), _indexCatalogEntry(ice), _descriptor(ice->descriptor()) {}
+    : _store(std::move(store)),
+      _indexCatalogEntry(ice),
+      _descriptor(ice->descriptor()),
+      _keyGen(_descriptor->keyPattern(), _descriptor->pathProjection()) {
+    // Normalize the 'columnstoreProjection' index option to facilitate its comparison as part of
+    // index signature.
+    if (!_descriptor->pathProjection().isEmpty()) {
+        auto* projExec = getColumnstoreProjection()->exec();
+        ice->descriptor()->_setNormalizedPathProjection(
+            projExec->serializeTransformation(boost::none).toBson());
+    }
+}
 
 class ColumnStoreAccessMethod::BulkBuilder final : public IndexAccessMethod::BulkBuilder {
 public:
diff --git a/src/mongo/db/index/columns_access_method.h b/src/mongo/db/index/columns_access_method.h
index f9ef8699c40..91a66ca4503 100644
--- a/src/mongo/db/index/columns_access_method.h
+++ b/src/mongo/db/index/columns_access_method.h
@@ -30,6 +30,7 @@
 #pragma once
 
 #include "mongo/db/index/column_cell.h"
+#include "mongo/db/index/column_key_generator.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/storage/column_store.h"
 
@@ -47,6 +48,13 @@ public:
 
     ColumnStoreAccessMethod(IndexCatalogEntry* ice, std::unique_ptr<ColumnStore>);
 
+    /**
+     * Returns a pointer to the ColumnstoreProjection owned by the underlying ColumnKeyGenerator.
+     */
+    const ColumnStoreProjection* getColumnstoreProjection() const {
+        return _keyGen.getColumnstoreProjection();
+    }
+
 
     //
     // Generic IndexAccessMethod APIs
@@ -111,5 +119,6 @@ private:
     const std::unique_ptr<ColumnStore> _store;
     IndexCatalogEntry* const _indexCatalogEntry;  // owned by IndexCatalog
     const IndexDescriptor* const _descriptor;
+    const ColumnKeyGenerator _keyGen;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/index/index_descriptor.cpp b/src/mongo/db/index/index_descriptor.cpp
index 2d33a32d84b..30198d1457c 100644
--- a/src/mongo/db/index/index_descriptor.cpp
+++ b/src/mongo/db/index/index_descriptor.cpp
@@ -55,20 +55,21 @@ std::map<StringData, BSONElement> populateOptionsMapForEqualityCheck(const BSONO
 
     // These index options are not considered for equality.
     static const StringDataSet kIndexOptionsNotConsideredForEqualityCheck{
-        IndexDescriptor::kKeyPatternFieldName,         // checked specially
-        IndexDescriptor::kNamespaceFieldName,          // removed in 4.4
-        IndexDescriptor::kIndexNameFieldName,          // checked separately
-        IndexDescriptor::kIndexVersionFieldName,       // not considered for equivalence
-        IndexDescriptor::kTextVersionFieldName,        // same as index version
-        IndexDescriptor::k2dsphereVersionFieldName,    // same as index version
-        IndexDescriptor::kBackgroundFieldName,         // this is a creation time option only
-        IndexDescriptor::kDropDuplicatesFieldName,     // this is now ignored
-        IndexDescriptor::kHiddenFieldName,             // not considered for equivalence
-        IndexDescriptor::kCollationFieldName,          // checked specially
-        IndexDescriptor::kPartialFilterExprFieldName,  // checked specially
-        IndexDescriptor::kUniqueFieldName,             // checked specially
-        IndexDescriptor::kSparseFieldName,             // checked specially
-        IndexDescriptor::kPathProjectionFieldName,     // checked specially
+        IndexDescriptor::kKeyPatternFieldName,             // checked specially
+        IndexDescriptor::kNamespaceFieldName,              // removed in 4.4
+        IndexDescriptor::kIndexNameFieldName,              // checked separately
+        IndexDescriptor::kIndexVersionFieldName,           // not considered for equivalence
+        IndexDescriptor::kTextVersionFieldName,            // same as index version
+        IndexDescriptor::k2dsphereVersionFieldName,        // same as index version
+        IndexDescriptor::kBackgroundFieldName,             // this is a creation time option only
+        IndexDescriptor::kDropDuplicatesFieldName,         // this is now ignored
+        IndexDescriptor::kHiddenFieldName,                 // not considered for equivalence
+        IndexDescriptor::kCollationFieldName,              // checked specially
+        IndexDescriptor::kPartialFilterExprFieldName,      // checked specially
+        IndexDescriptor::kUniqueFieldName,                 // checked specially
+        IndexDescriptor::kSparseFieldName,                 // checked specially
+        IndexDescriptor::kWildcardProjectionFieldName,     // checked specially
+        IndexDescriptor::kColumnStoreProjectionFieldName,  // checked specially
     };
 
     BSONObjIterator it(spec);
@@ -102,7 +103,8 @@ constexpr StringData IndexDescriptor::kKeyPatternFieldName;
 constexpr StringData IndexDescriptor::kLanguageOverrideFieldName;
 constexpr StringData IndexDescriptor::kNamespaceFieldName;
 constexpr StringData IndexDescriptor::kPartialFilterExprFieldName;
-constexpr StringData IndexDescriptor::kPathProjectionFieldName;
+constexpr StringData IndexDescriptor::kWildcardProjectionFieldName;
+constexpr StringData IndexDescriptor::kColumnStoreProjectionFieldName;
 constexpr StringData IndexDescriptor::kSparseFieldName;
 constexpr StringData IndexDescriptor::kStorageEngineFieldName;
 constexpr StringData IndexDescriptor::kTextVersionFieldName;
@@ -117,7 +119,7 @@ IndexDescriptor::IndexDescriptor(const std::string& accessMethodName, BSONObj in
       _infoObj(infoObj.getOwned()),
       _numFields(infoObj.getObjectField(IndexDescriptor::kKeyPatternFieldName).nFields()),
       _keyPattern(infoObj.getObjectField(IndexDescriptor::kKeyPatternFieldName).getOwned()),
-      _projection(infoObj.getObjectField(IndexDescriptor::kPathProjectionFieldName).getOwned()),
+      _projection(createPathProjection(infoObj)),
       _indexName(infoObj.getStringField(IndexDescriptor::kIndexNameFieldName)),
       _isIdIndex(isIdIndexPattern(_keyPattern)),
       _sparse(infoObj[IndexDescriptor::kSparseFieldName].trueValue()),
diff --git a/src/mongo/db/index/index_descriptor.h b/src/mongo/db/index/index_descriptor.h
index 115fc8b1d17..680ce6db091 100644
--- a/src/mongo/db/index/index_descriptor.h
+++ b/src/mongo/db/index/index_descriptor.h
@@ -81,7 +81,8 @@ public:
     static constexpr StringData kLanguageOverrideFieldName = "language_override"_sd;
     static constexpr StringData kNamespaceFieldName = "ns"_sd;  // Removed in 4.4
     static constexpr StringData kPartialFilterExprFieldName = "partialFilterExpression"_sd;
-    static constexpr StringData kPathProjectionFieldName = "wildcardProjection"_sd;
+    static constexpr StringData kWildcardProjectionFieldName = "wildcardProjection"_sd;
+    static constexpr StringData kColumnStoreProjectionFieldName = "columnstoreProjection"_sd;
     static constexpr StringData kSparseFieldName = "sparse"_sd;
     static constexpr StringData kStorageEngineFieldName = "storageEngine"_sd;
     static constexpr StringData kTextVersionFieldName = "textIndexVersion"_sd;
@@ -252,12 +253,28 @@ public:
     }
 
 private:
-    // This method should only ever be called by WildcardAccessMethod, to set the
-    // '_normalizedProjection' for descriptors associated with an existing IndexCatalogEntry.
+    // This method should only ever be called by WildcardAccessMethod or ColumnstoreAccessMethod, to
+    // set the '_normalizedProjection' for descriptors associated with an existing
+    // IndexCatalogEntry.
     void _setNormalizedPathProjection(BSONObj&& proj) {
         _normalizedProjection = std::move(proj);
     }
 
+    /**
+     * Returns wildcardProjection or columnstoreProjection projection
+     */
+    BSONObj createPathProjection(const BSONObj& infoObj) const {
+        if (const auto wildcardProjection =
+                infoObj[IndexDescriptor::kWildcardProjectionFieldName]) {
+            return wildcardProjection.Obj().getOwned();
+        } else if (const auto columnStoreProjection =
+                       infoObj[IndexDescriptor::kColumnStoreProjectionFieldName]) {
+            return columnStoreProjection.Obj().getOwned();
+        } else {
+            return BSONObj();
+        }
+    }
+
     // What access method should we use for this index?
     std::string _accessMethodName;
 
@@ -291,6 +308,7 @@ private:
     friend class IndexCatalogEntryImpl;
     friend class IndexCatalogEntryContainer;
     friend class WildcardAccessMethod;
+    friend class ColumnStoreAccessMethod;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/index/wildcard_key_generator.h b/src/mongo/db/index/wildcard_key_generator.h
index 4164fa9aa92..151c701e14a 100644
--- a/src/mongo/db/index/wildcard_key_generator.h
+++ b/src/mongo/db/index/wildcard_key_generator.h
@@ -29,7 +29,7 @@
 
 #pragma once
 
-#include "mongo/db/exec/wildcard_projection.h"
+#include "mongo/db/exec/index_path_projection.h"
 #include "mongo/db/field_ref.h"
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/storage/key_string.h"
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index 3a631c57988..205f9badbd5 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -3050,23 +3050,43 @@ std::vector<BSONObj> IndexBuildsCoordinator::normalizeIndexSpecs(
     // not impact our internal index comparison semantics, since we compare based on the parsed
     // MatchExpression trees rather than the serialized BSON specs. See SERVER-54357.
 
-    // If any of the specs describe wildcard indexes, normalize the wildcard projections if present.
-    // This will change all specs of the form {"a.b.c": 1} to normalized form {a: {b: {c : 1}}}.
+    // If any of the specs describe wildcard or columnstore indexes, normalize the respective
+    // projections if present. This will change all specs of the form {"a.b.c": 1} to normalized
+    // form {a: {b: {c : 1}}}.
     std::transform(normalSpecs.begin(), normalSpecs.end(), normalSpecs.begin(), [](auto& spec) {
-        const auto kProjectionName = IndexDescriptor::kPathProjectionFieldName;
-        const auto pathProjectionSpec = spec.getObjectField(kProjectionName);
-        static const auto kWildcardKeyPattern = BSON("$**" << 1);
-        // It's illegal for the user to explicitly specify an empty wildcardProjection for creating
-        // a {"$**":1} index, and specify any wildcardProjection for a {"field.$**": 1} index. If
-        // the projection is empty, then it means that there is no projection to normalize.
-        if (pathProjectionSpec.isEmpty()) {
+        BSONObj pathProjectionSpec;
+        bool isWildcard = false;
+        auto wildcardProjection = spec[IndexDescriptor::kWildcardProjectionFieldName];
+        auto columnStoreProjection = spec[IndexDescriptor::kColumnStoreProjectionFieldName];
+        if (wildcardProjection) {
+            pathProjectionSpec = wildcardProjection.Obj();
+            invariant(!spec[IndexDescriptor::kColumnStoreProjectionFieldName]);
+            isWildcard = true;
+        } else if (columnStoreProjection) {
+            pathProjectionSpec = columnStoreProjection.Obj();
+            invariant(!spec[IndexDescriptor::kWildcardProjectionFieldName]);
+        } else {
+            // No projection to normalize.
             return spec;
         }
-        auto wildcardProjection =
-            WildcardKeyGenerator::createProjectionExecutor(kWildcardKeyPattern, pathProjectionSpec);
+        uassert(ErrorCodes::InvalidIndexSpecificationOption,
+                "Can't enable both wildcardProjection and columnstoreProjection",
+                !(wildcardProjection && columnStoreProjection));
+
+        // Exactly one of wildcardProjection or columnstoreProjection is enabled
+        const auto projectionName = isWildcard ? IndexDescriptor::kWildcardProjectionFieldName
+                                               : IndexDescriptor::kColumnStoreProjectionFieldName;
+        static const auto kFieldSetKeyPattern = isWildcard ? BSON("$**" << 1)
+                                                           : BSON("$**"
+                                                                  << "columnstore");
+        auto indexPathProjection = isWildcard
+            ? static_cast<IndexPathProjection>(WildcardKeyGenerator::createProjectionExecutor(
+                  kFieldSetKeyPattern, pathProjectionSpec))
+            : static_cast<IndexPathProjection>(ColumnKeyGenerator::createProjectionExecutor(
+                  kFieldSetKeyPattern, pathProjectionSpec));
         auto normalizedProjection =
-            wildcardProjection.exec()->serializeTransformation(boost::none).toBson();
-        return spec.addField(BSON(kProjectionName << normalizedProjection).firstElement());
+            indexPathProjection.exec()->serializeTransformation(boost::none).toBson();
+        return spec.addField(BSON(projectionName << normalizedProjection).firstElement());
     });
     return normalSpecs;
 }
diff --git a/src/mongo/db/index_builds_coordinator.h b/src/mongo/db/index_builds_coordinator.h
index bb1e372ae8f..5a6803a0e78 100644
--- a/src/mongo/db/index_builds_coordinator.h
+++ b/src/mongo/db/index_builds_coordinator.h
@@ -44,6 +44,7 @@
 #include "mongo/db/catalog/index_builds_manager.h"
 #include "mongo/db/commands/server_status.h"
 #include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/index/column_key_generator.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/rebuild_indexes.h"
 #include "mongo/db/repl/oplog_entry.h"
diff --git a/src/mongo/db/index_names.cpp b/src/mongo/db/index_names.cpp
index 97a915e2226..9f9af217bba 100644
--- a/src/mongo/db/index_names.cpp
+++ b/src/mongo/db/index_names.cpp
@@ -67,7 +67,12 @@ string IndexNames::findPluginName(const BSONObj& keyPattern) {
         if (String == e.type()) {
             return e.String();
         } else if ((fieldName == "$**") || fieldName.endsWith(".$**")) {
-            return IndexNames::WILDCARD;
+            if (keyPattern.firstElement().type() == String &&
+                keyPattern.firstElement().fieldNameStringData() == "columnstore"_sd) {
+                return IndexNames::COLUMN;
+            } else {
+                return IndexNames::WILDCARD;
+            }
         } else
             continue;
     }
diff --git a/src/mongo/db/list_indexes.idl b/src/mongo/db/list_indexes.idl
index 44c7aeffd03..4cafe6e3e99 100644
--- a/src/mongo/db/list_indexes.idl
+++ b/src/mongo/db/list_indexes.idl
@@ -40,7 +40,7 @@ imports:
 structs:
     ListIndexesReplyItem:
         description: "An item in the listIndexes command's reply"
-        # If adding new fields, the set 'allowedFieldsNames' in 
+        # If adding new fields, the set 'allowedFieldsNames' in
         # 'src/mongo/db/commands/list_indexes.cpp' should also be updated.
         fields:
             #
@@ -138,6 +138,10 @@ structs:
                 type: object_owned
                 optional: true
                 unstable: false
+            columnstoreProjection:
+                type: object_owned
+                optional: true
+                unstable: true
             coarsestIndexedLevel:
                 type: safeInt
                 optional: true
@@ -164,7 +168,7 @@ structs:
                 unstable: true
             #
             # Depending on the values of includeIndexBuildInfo and includeBuildUUIDs, indexes may
-            # appear with a combination of these three fields. Specifically, if includeIndexBuildInfo 
+            # appear with a combination of these three fields. Specifically, if includeIndexBuildInfo
             # is set, completed index builds will appear with spec, and in-progress index builds will
             # appear with both spec and indexBuildInfo, while if includeBuildUUIDs is set, in-progress
             # index builds will appear with both spec and buildUUID. They're required, but marked
@@ -202,7 +206,7 @@ structs:
     ListIndexesReply:
         description: "The listIndexes command's reply."
         fields:
-            cursor: 
+            cursor:
                 type: ListIndexesReplyCursor
                 unstable: false
             # Included so mongos can parse shards' listIndexes replies.
diff --git a/src/mongo/db/query/get_executor_test.cpp b/src/mongo/db/query/get_executor_test.cpp
index aca45d23026..431430259e8 100644
--- a/src/mongo/db/query/get_executor_test.cpp
+++ b/src/mongo/db/query/get_executor_test.cpp
@@ -37,9 +37,9 @@
 #include <string>
 
 #include "mongo/bson/simple_bsonobj_comparator.h"
+#include "mongo/db/exec/index_path_projection.h"
 #include "mongo/db/exec/projection_executor.h"
 #include "mongo/db/exec/projection_executor_builder.h"
-#include "mongo/db/exec/wildcard_projection.h"
 #include "mongo/db/json.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/query/projection_parser.h"
diff --git a/src/mongo/db/query/index_entry.h b/src/mongo/db/query/index_entry.h
index 6f69d388c9d..40313fdfed6 100644
--- a/src/mongo/db/query/index_entry.h
+++ b/src/mongo/db/query/index_entry.h
@@ -43,7 +43,8 @@
 namespace mongo {
 class CollatorInterface;
 class MatchExpression;
-class WildcardProjection;
+class IndexPathProjection;
+using WildcardProjection = IndexPathProjection;
 
 /**
  * A CoreIndexInfo is a representation of an index in the catalog with parsed information which is
@@ -69,6 +70,7 @@ struct CoreIndexInfo {
           collator(ci),
           wildcardProjection(wildcardProj) {
         // We always expect a projection executor for $** indexes, and none otherwise.
+        // TODO SERVER-67140: Add columnstoreProjection and invariant
         invariant((type == IndexType::INDEX_WILDCARD) == (wildcardProjection != nullptr));
     }
 
@@ -262,7 +264,7 @@ struct ColumnIndexEntry {
 
     std::string catalogName;
 
-    // TODO SERVER-63123: Projection, probably need some kind of disambiguator.
+    // TODO SERVER-67140: Projection, probably need some kind of disambiguator.
 };
 
 std::ostream& operator<<(std::ostream& stream, const IndexEntry::Identifier& ident);
diff --git a/src/mongo/db/query/projection_parser.cpp b/src/mongo/db/query/projection_parser.cpp
index 7110775621d..e093d840d9f 100644
--- a/src/mongo/db/query/projection_parser.cpp
+++ b/src/mongo/db/query/projection_parser.cpp
@@ -592,7 +592,7 @@ Projection parseAndAnalyze(boost::intrusive_ptr<ExpressionContext> expCtx,
                            const BSONObj& queryObj,
                            ProjectionPolicies policies,
                            bool shouldOptimize) {
-    if (!policies.findOnlyFeaturesAllowed()) {
+    if (!policies.emptyProjectionAllowed()) {
         // In agg-style syntax it is illegal to have an empty projection specification.
         uassert(51272, "projection specification must have at least one field", !obj.isEmpty());
     }
diff --git a/src/mongo/db/query/projection_policies.h b/src/mongo/db/query/projection_policies.h
index 09094227c86..1e9214f2515 100644
--- a/src/mongo/db/query/projection_policies.h
+++ b/src/mongo/db/query/projection_policies.h
@@ -55,6 +55,9 @@ struct ProjectionPolicies {
     // Whether $elemMatch, find() $slice and positional projection are allowed.
     enum class FindOnlyFeaturesPolicy { kBanFindOnlyFeatures, kAllowFindOnlyFeatures };
 
+    // Whether the empty projection, {}, is permitted.
+    enum class EmptyProjectionPolicy { kBanEmptyProjection, kAllowEmptyProjection };
+
     static const DefaultIdPolicy kDefaultIdPolicyDefault = DefaultIdPolicy::kIncludeId;
     static const ArrayRecursionPolicy kArrayRecursionPolicyDefault =
         ArrayRecursionPolicy::kRecurseNestedArrays;
@@ -62,46 +65,66 @@ struct ProjectionPolicies {
         ComputedFieldsPolicy::kAllowComputedFields;
     static const FindOnlyFeaturesPolicy kFindOnlyFeaturesPolicyDefault =
         FindOnlyFeaturesPolicy::kBanFindOnlyFeatures;
+    static const EmptyProjectionPolicy kEmptyProjectionPolicyDefault =
+        EmptyProjectionPolicy::kBanEmptyProjection;
 
     static ProjectionPolicies findProjectionPolicies() {
         return ProjectionPolicies{kDefaultIdPolicyDefault,
                                   kArrayRecursionPolicyDefault,
                                   kComputedFieldsPolicyDefault,
-                                  FindOnlyFeaturesPolicy::kAllowFindOnlyFeatures};
+                                  FindOnlyFeaturesPolicy::kAllowFindOnlyFeatures,
+                                  EmptyProjectionPolicy::kAllowEmptyProjection};
     }
 
     static ProjectionPolicies aggregateProjectionPolicies() {
         return ProjectionPolicies{kDefaultIdPolicyDefault,
                                   kArrayRecursionPolicyDefault,
                                   kComputedFieldsPolicyDefault,
-                                  FindOnlyFeaturesPolicy::kBanFindOnlyFeatures};
+                                  FindOnlyFeaturesPolicy::kBanFindOnlyFeatures,
+                                  kEmptyProjectionPolicyDefault};
     }
 
     static ProjectionPolicies wildcardIndexSpecProjectionPolicies() {
         return ProjectionPolicies{DefaultIdPolicy::kExcludeId,
                                   ArrayRecursionPolicy::kDoNotRecurseNestedArrays,
                                   ComputedFieldsPolicy::kBanComputedFields,
-                                  FindOnlyFeaturesPolicy::kBanFindOnlyFeatures};
+                                  FindOnlyFeaturesPolicy::kBanFindOnlyFeatures,
+                                  kEmptyProjectionPolicyDefault};
+    }
+
+    static ProjectionPolicies columnStoreIndexSpecProjectionPolicies() {
+        return ProjectionPolicies{kDefaultIdPolicyDefault,
+                                  ArrayRecursionPolicy::kDoNotRecurseNestedArrays,
+                                  ComputedFieldsPolicy::kBanComputedFields,
+                                  FindOnlyFeaturesPolicy::kBanFindOnlyFeatures,
+                                  EmptyProjectionPolicy::kAllowEmptyProjection};
     }
 
     ProjectionPolicies(
         DefaultIdPolicy idPolicy = kDefaultIdPolicyDefault,
         ArrayRecursionPolicy arrayRecursionPolicy = kArrayRecursionPolicyDefault,
         ComputedFieldsPolicy computedFieldsPolicy = kComputedFieldsPolicyDefault,
-        FindOnlyFeaturesPolicy findOnlyFeaturesPolicy = kFindOnlyFeaturesPolicyDefault)
+        FindOnlyFeaturesPolicy findOnlyFeaturesPolicy = kFindOnlyFeaturesPolicyDefault,
+        EmptyProjectionPolicy emptyProjectionPolicy = kEmptyProjectionPolicyDefault)
         : idPolicy(idPolicy),
           arrayRecursionPolicy(arrayRecursionPolicy),
           computedFieldsPolicy(computedFieldsPolicy),
-          findOnlyFeaturesPolicy(findOnlyFeaturesPolicy) {}
+          findOnlyFeaturesPolicy(findOnlyFeaturesPolicy),
+          emptyProjectionPolicy(emptyProjectionPolicy) {}
 
     const DefaultIdPolicy idPolicy;
     const ArrayRecursionPolicy arrayRecursionPolicy;
     const ComputedFieldsPolicy computedFieldsPolicy;
     const FindOnlyFeaturesPolicy findOnlyFeaturesPolicy;
+    const EmptyProjectionPolicy emptyProjectionPolicy;
 
     bool findOnlyFeaturesAllowed() const {
         return findOnlyFeaturesPolicy == FindOnlyFeaturesPolicy::kAllowFindOnlyFeatures;
     }
+
+    bool emptyProjectionAllowed() const {
+        return emptyProjectionPolicy == EmptyProjectionPolicy::kAllowEmptyProjection;
+    }
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/query/query_planner.cpp b/src/mongo/db/query/query_planner.cpp
index 6f37288e733..83cc6617d44 100644
--- a/src/mongo/db/query/query_planner.cpp
+++ b/src/mongo/db/query/query_planner.cpp
@@ -218,7 +218,7 @@ void tryToAddColumnScan(const QueryPlannerParams& params,
         return;
     }
 
-    // TODO SERVER-63123: Check if the columnar index actually provides the fields we need.
+    // TODO SERVER-67140: Check if the columnar index actually provides the fields we need.
     std::unique_ptr<MatchExpression> residualPredicate;
     StringMap<std::unique_ptr<MatchExpression>> filterSplitByColumn;
     if (params.options & QueryPlannerParams::GENERATE_PER_COLUMN_FILTERS) {
diff --git a/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp b/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
index 92ac4c82c0b..893252abec3 100644
--- a/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
+++ b/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
@@ -199,7 +199,7 @@ protected:
         BSONObjBuilder bob = std::move(BSONObjBuilder() << "name" << name << "key" << key);
 
         if (!pathProjection.isEmpty())
-            bob << IndexDescriptor::kPathProjectionFieldName << pathProjection;
+            bob << IndexDescriptor::kWildcardProjectionFieldName << pathProjection;
 
         auto indexSpec = (bob << "v" << kIndexVersion << "background" << background).obj();