diff --git a/go/libraries/doltcore/sqle/enginetest/dolt_queries.go b/go/libraries/doltcore/sqle/enginetest/dolt_queries.go
index 346a423c31..70335711c6 100644
--- a/go/libraries/doltcore/sqle/enginetest/dolt_queries.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_queries.go
@@ -1962,6 +1962,13 @@ var HistorySystemTableScriptTests = []queries.ScriptTest{
 				Query:    "select pk, c2 from dolt_history_t where commit_hash=@Commit2 order by pk;",
 				Expected: []sql.Row{{1, 3}, {4, 6}},
 			},
+			{
+				// When filtering on a column from the original table, we use the primary index here, but because
+				// column tags have changed in previous versions of the table, the index tags don't match up completely.
+				// https://github.com/dolthub/dolt/issues/6891
+				Query:    "select pk, c1, c2 from dolt_history_t where pk=4;",
+				Expected: []sql.Row{{4, 5, 6}},
+			},
 		},
 	},
 	{
diff --git a/go/libraries/doltcore/sqle/index/index_lookup.go b/go/libraries/doltcore/sqle/index/index_lookup.go
index f5b26958ea..ea7e02464e 100644
--- a/go/libraries/doltcore/sqle/index/index_lookup.go
+++ b/go/libraries/doltcore/sqle/index/index_lookup.go
@@ -281,8 +281,13 @@ func NewLookupBuilder(
 		}, nil
 	case idx.coversColumns(s, projections):
 		return newCoveringLookupBuilder(base), nil
+	case idx.ID() == "PRIMARY":
+		// If we are using the primary index, always use a covering lookup builder. In some cases, coversColumns
+		// can return false, for example if a column was modified in an older version and has a different tag than
+		// the current schema. In those cases, the primary index is still the best we have, so go ahead and use it.
+		return newCoveringLookupBuilder(base), nil
 	default:
-		return newNonCoveringLookupBuilder(s, base), nil
+		return newNonCoveringLookupBuilder(s, base)
 	}
 }
 
@@ -301,7 +306,17 @@ func newCoveringLookupBuilder(b *baseLookupBuilder) *coveringLookupBuilder {
 	}
 }
 
-func newNonCoveringLookupBuilder(s *durableIndexState, b *baseLookupBuilder) *nonCoveringLookupBuilder {
+// newNonCoveringLookupBuilder returns a LookupBuilder that uses the specified index state and
+// base lookup builder to create a nonCoveringLookupBuilder that uses the secondary index (from
+// |b|) to find the PK row identifier, and then uses that PK to look up the complete row from
+// the primary index (from |s|). If a baseLookupBuilder built on the primary index is passed in,
+// this function returns an error.
+func newNonCoveringLookupBuilder(s *durableIndexState, b *baseLookupBuilder) (*nonCoveringLookupBuilder, error) {
+	if b.idx.ID() == "PRIMARY" {
+		return nil, fmt.Errorf("incompatible index passed to newNonCoveringLookupBuilder: " +
+			"primary index passed, but only secondary indexes are supported")
+	}
+
 	primary := durable.ProllyMapFromIndex(s.Primary)
 	priKd, _ := primary.Descriptors()
 	tbBld := val.NewTupleBuilder(priKd)
@@ -316,7 +331,7 @@ func newNonCoveringLookupBuilder(s *durableIndexState, b *baseLookupBuilder) *no
 		keyMap:            keyProj,
 		valMap:            valProj,
 		ordMap:            ordProj,
-	}
+	}, nil
 }
 
 var _ LookupBuilder = (*baseLookupBuilder)(nil)
@@ -411,7 +426,7 @@ func (lb *coveringLookupBuilder) NewRowIter(ctx *sql.Context, part sql.Partition
 
 // nonCoveringLookupBuilder constructs row iters for non-covering lookups,
 // where we need to seek on the secondary table for key identity, and then
-// the primary table to fill all requrested projections.
+// the primary table to fill all requested projections.
 type nonCoveringLookupBuilder struct {
 	*baseLookupBuilder