diff --git a/clients/shared/merge.go b/clients/shared/merge.go
index 0621d00ef..7b3f06927 100644
--- a/clients/shared/merge.go
+++ b/clients/shared/merge.go
@@ -93,7 +93,7 @@ func Merge(dwh destination.DataWarehouse, tableData *optimization.TableData, cfg
 		for attempts := 0; attempts < backfillMaxRetries; attempts++ {
 			backfillErr = BackfillColumn(cfg, dwh, col, tableID)
 			if backfillErr == nil {
-				tableConfig.Columns().UpsertColumn(col.RawName(), columns.UpsertColumnArg{
+				tableConfig.Columns().UpsertColumn(col.Name(), columns.UpsertColumnArg{
 					Backfilled: ptr.ToBool(true),
 				})
 				break
@@ -110,7 +110,7 @@ func Merge(dwh destination.DataWarehouse, tableData *optimization.TableData, cfg
 		}
 
 		if backfillErr != nil {
-			return fmt.Errorf("failed to backfill col: %s, default value: %v, err: %w", col.RawName(), col.RawDefaultValue(), backfillErr)
+			return fmt.Errorf("failed to backfill col: %s, default value: %v, err: %w", col.Name(), col.RawDefaultValue(), backfillErr)
 		}
 	}
 
diff --git a/clients/shared/utils.go b/clients/shared/utils.go
index 606f87ce1..64f9faa82 100644
--- a/clients/shared/utils.go
+++ b/clients/shared/utils.go
@@ -45,7 +45,7 @@ func BackfillColumn(cfg config.Config, dwh destination.DataWarehouse, column col
 		tableID.FullyQualifiedName(), escapedCol, defaultVal, additionalEscapedCol,
 	)
 	slog.Info("Backfilling column",
-		slog.String("colName", column.RawName()),
+		slog.String("colName", column.Name()),
 		slog.String("query", query),
 		slog.String("table", tableID.FullyQualifiedName()),
 	)
diff --git a/clients/snowflake/ddl_test.go b/clients/snowflake/ddl_test.go
index c2e8a3e77..61c607576 100644
--- a/clients/snowflake/ddl_test.go
+++ b/clients/snowflake/ddl_test.go
@@ -41,7 +41,7 @@ func (s *SnowflakeTestSuite) TestMutateColumnsWithMemoryCacheDeletions() {
 	nameCol := columns.NewColumn("name", typing.String)
 	tc := s.stageStore.configMap.TableConfig(tableID)
 
-	val := tc.ShouldDeleteColumn(nameCol.RawName(), time.Now().Add(-1*6*time.Hour), true)
+	val := tc.ShouldDeleteColumn(nameCol.Name(), time.Now().Add(-1*6*time.Hour), true)
 	assert.False(s.T(), val, "should not try to delete this column")
 	assert.Equal(s.T(), len(s.stageStore.configMap.TableConfig(tableID).ReadOnlyColumnsToDelete()), 1)
 
@@ -68,23 +68,23 @@ func (s *SnowflakeTestSuite) TestShouldDeleteColumn() {
 
 	nameCol := columns.NewColumn("name", typing.String)
 	// Let's try to delete name.
-	allowed := s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.RawName(),
+	allowed := s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.Name(),
 		time.Now().Add(-1*(6*time.Hour)), true)
 
 	assert.Equal(s.T(), allowed, false, "should not be allowed to delete")
 
 	// Process tried to delete, but it's lagged.
-	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.RawName(),
+	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.Name(),
 		time.Now().Add(-1*(6*time.Hour)), true)
 
 	assert.Equal(s.T(), allowed, false, "should not be allowed to delete")
 
 	// Process now caught up, and is asking if we can delete, should still be no.
-	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.RawName(), time.Now(), true)
+	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.Name(), time.Now(), true)
 	assert.Equal(s.T(), allowed, false, "should not be allowed to delete still")
 
 	// Process is finally ahead, has permission to delete now.
-	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.RawName(),
+	allowed = s.stageStore.configMap.TableConfig(tableID).ShouldDeleteColumn(nameCol.Name(),
 		time.Now().Add(2*constants.DeletionConfidencePadding), true)
 
 	assert.Equal(s.T(), allowed, true, "should now be allowed to delete")
diff --git a/lib/cdc/mysql/debezium_test.go b/lib/cdc/mysql/debezium_test.go
index 783ab83c8..49f6fb760 100644
--- a/lib/cdc/mysql/debezium_test.go
+++ b/lib/cdc/mysql/debezium_test.go
@@ -353,7 +353,7 @@ func (m *MySQLTestSuite) TestGetEventFromBytes() {
 
 	col, isOk := cols.GetColumn("abcdef")
 	assert.True(m.T(), isOk)
-	assert.Equal(m.T(), "abcdef", col.RawName())
+	assert.Equal(m.T(), "abcdef", col.Name())
 	for key := range evtData {
 		if strings.Contains(key, constants.ArtiePrefix) {
 			continue
@@ -361,6 +361,6 @@ func (m *MySQLTestSuite) TestGetEventFromBytes() {
 
 		col, isOk = cols.GetColumn(strings.ToLower(key))
 		assert.Equal(m.T(), true, isOk, key)
-		assert.Equal(m.T(), typing.Invalid, col.KindDetails, fmt.Sprintf("colName: %v, evtData key: %v", col.RawName(), key))
+		assert.Equal(m.T(), typing.Invalid, col.KindDetails, fmt.Sprintf("colName: %v, evtData key: %v", col.Name(), key))
 	}
 }
diff --git a/lib/cdc/util/relational_event_test.go b/lib/cdc/util/relational_event_test.go
index e5a421ed7..72d903864 100644
--- a/lib/cdc/util/relational_event_test.go
+++ b/lib/cdc/util/relational_event_test.go
@@ -74,8 +74,8 @@ func TestSource_GetOptionalSchema(t *testing.T) {
 
 	for _, _col := range cols.GetColumns() {
 		// All the other columns do not have a default value.
-		if _col.RawName() != "boolean_column" {
-			assert.Nil(t, _col.RawDefaultValue(), _col.RawName())
+		if _col.Name() != "boolean_column" {
+			assert.Nil(t, _col.RawDefaultValue(), _col.Name())
 		}
 	}
 }
diff --git a/lib/destination/ddl/ddl.go b/lib/destination/ddl/ddl.go
index 886352527..c8f0b864b 100644
--- a/lib/destination/ddl/ddl.go
+++ b/lib/destination/ddl/ddl.go
@@ -91,7 +91,7 @@ func (a AlterTableArgs) AlterTable(cols ...columns.Column) error {
 		}
 
 		if a.ColumnOp == constants.Delete {
-			if !a.Tc.ShouldDeleteColumn(col.RawName(), a.CdcTime, a.ContainOtherOperations) {
+			if !a.Tc.ShouldDeleteColumn(col.Name(), a.CdcTime, a.ContainOtherOperations) {
 				continue
 			}
 		}
diff --git a/lib/destination/ddl/ddl_bq_test.go b/lib/destination/ddl/ddl_bq_test.go
index fe4a47448..123fcc86c 100644
--- a/lib/destination/ddl/ddl_bq_test.go
+++ b/lib/destination/ddl/ddl_bq_test.go
@@ -152,10 +152,10 @@ func (d *DDLTestSuite) TestAlterTableAddColumns() {
 	assert.Equal(d.T(), newColsLen+existingColsLen, len(d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns().GetColumns()), d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns())
 	// Check by iterating over the columns
 	for _, column := range d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns().GetColumns() {
-		existingCol, isOk := existingCols.GetColumn(column.RawName())
+		existingCol, isOk := existingCols.GetColumn(column.Name())
 		if !isOk {
 			// Check new cols?
-			existingCol.KindDetails, isOk = newCols[column.RawName()]
+			existingCol.KindDetails, isOk = newCols[column.Name()]
 		}
 
 		assert.True(d.T(), isOk)
@@ -211,7 +211,7 @@ func (d *DDLTestSuite) TestAlterTableAddColumnsSomeAlreadyExist() {
 	assert.Equal(d.T(), existingColsLen, len(d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns().GetColumns()), d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns())
 	// Check by iterating over the columns
 	for _, column := range d.bigQueryStore.GetConfigMap().TableConfig(tableID).Columns().GetColumns() {
-		existingCol, isOk := existingCols.GetColumn(column.RawName())
+		existingCol, isOk := existingCols.GetColumn(column.Name())
 		assert.True(d.T(), isOk)
 		assert.Equal(d.T(), column.KindDetails, existingCol.KindDetails)
 	}
diff --git a/lib/destination/ddl/ddl_sflk_test.go b/lib/destination/ddl/ddl_sflk_test.go
index 65f9926b2..352efba2a 100644
--- a/lib/destination/ddl/ddl_sflk_test.go
+++ b/lib/destination/ddl/ddl_sflk_test.go
@@ -108,15 +108,15 @@ func (d *DDLTestSuite) TestAlterTableAdd() {
 	for _, column := range tableConfig.Columns().GetColumns() {
 		var found bool
 		for _, expCol := range cols {
-			if found = column.RawName() == expCol.RawName(); found {
-				assert.Equal(d.T(), column.KindDetails, expCol.KindDetails, fmt.Sprintf("wrong col kind, col: %s", column.RawName()))
+			if found = column.Name() == expCol.Name(); found {
+				assert.Equal(d.T(), column.KindDetails, expCol.KindDetails, fmt.Sprintf("wrong col kind, col: %s", column.Name()))
 				break
 			}
 		}
 
 		assert.True(d.T(), found,
 			fmt.Sprintf("Col not found: %s, actual list: %v, expected list: %v",
-				column.RawName(), tableConfig.Columns(), cols))
+				column.Name(), tableConfig.Columns(), cols))
 	}
 }
 
@@ -150,7 +150,7 @@ func (d *DDLTestSuite) TestAlterTableDeleteDryRun() {
 	for col := range tableConfig.ReadOnlyColumnsToDelete() {
 		var found bool
 		for _, expCol := range cols {
-			if found = col == expCol.RawName(); found {
+			if found = col == expCol.Name(); found {
 				break
 			}
 		}
@@ -161,7 +161,7 @@ func (d *DDLTestSuite) TestAlterTableDeleteDryRun() {
 	}
 
 	for i := 0; i < len(cols); i++ {
-		colToActuallyDelete := cols[i].RawName()
+		colToActuallyDelete := cols[i].Name()
 		// Now let's check the timestamp
 		assert.True(d.T(), tableConfig.ReadOnlyColumnsToDelete()[colToActuallyDelete].After(time.Now()))
 		// Now let's actually try to dial the time back, and it should actually try to delete.
@@ -214,7 +214,7 @@ func (d *DDLTestSuite) TestAlterTableDelete() {
 	for col := range tableConfig.ReadOnlyColumnsToDelete() {
 		var found bool
 		for _, expCol := range cols {
-			if found = col == expCol.RawName(); found {
+			if found = col == expCol.Name(); found {
 				break
 			}
 		}
diff --git a/lib/destination/dml/merge.go b/lib/destination/dml/merge.go
index c75e5579e..e40f1d890 100644
--- a/lib/destination/dml/merge.go
+++ b/lib/destination/dml/merge.go
@@ -208,9 +208,9 @@ func (m *MergeArgument) GetStatement() (string, error) {
 	for _, primaryKey := range m.PrimaryKeys {
 		// We'll need to escape the primary key as well.
 		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.EscapedName(m.Dialect), primaryKey.EscapedName(m.Dialect))
-		pkCol, isOk := m.Columns.GetColumn(primaryKey.RawName())
+		pkCol, isOk := m.Columns.GetColumn(primaryKey.Name())
 		if !isOk {
-			return "", fmt.Errorf("column: %s does not exist in columnToType: %v", primaryKey.RawName(), m.Columns)
+			return "", fmt.Errorf("column: %s does not exist in columnToType: %v", primaryKey.Name(), m.Columns)
 		}
 
 		if m.DestKind == constants.BigQuery && pkCol.KindDetails.Kind == typing.Struct.Kind {
diff --git a/lib/destination/types/table_config.go b/lib/destination/types/table_config.go
index 755f5b66c..79d6515a0 100644
--- a/lib/destination/types/table_config.go
+++ b/lib/destination/types/table_config.go
@@ -64,15 +64,15 @@ func (d *DwhTableConfig) MutateInMemoryColumns(createTable bool, columnOp consta
 		for _, col := range cols {
 			d.columns.AddColumn(col)
 			// Delete from the permissions table, if exists.
-			delete(d.columnsToDelete, col.RawName())
+			delete(d.columnsToDelete, col.Name())
 		}
 
 		d.createTable = createTable
 	case constants.Delete:
 		for _, col := range cols {
 			// Delete from the permissions and in-memory table
-			d.columns.DeleteColumn(col.RawName())
-			delete(d.columnsToDelete, col.RawName())
+			d.columns.DeleteColumn(col.Name())
+			delete(d.columnsToDelete, col.Name())
 		}
 	}
 }
@@ -91,7 +91,7 @@ func (d *DwhTableConfig) AuditColumnsToDelete(colsToDelete []columns.Column) {
 	for colName := range d.columnsToDelete {
 		var found bool
 		for _, col := range colsToDelete {
-			if found = col.RawName() == colName; found {
+			if found = col.Name() == colName; found {
 				break
 			}
 		}
diff --git a/lib/optimization/event_update_test.go b/lib/optimization/event_update_test.go
index 2d5fa5e78..a61b04e2e 100644
--- a/lib/optimization/event_update_test.go
+++ b/lib/optimization/event_update_test.go
@@ -74,16 +74,16 @@ func TestTableData_UpdateInMemoryColumnsFromDestination(t *testing.T) {
 
 		// Testing backfill
 		for _, inMemoryCol := range tableData.inMemoryColumns.GetColumns() {
-			assert.False(t, inMemoryCol.Backfilled(), inMemoryCol.RawName())
+			assert.False(t, inMemoryCol.Backfilled(), inMemoryCol.Name())
 		}
 		backfilledCol := columns.NewColumn("bool_backfill", typing.Boolean)
 		backfilledCol.SetBackfilled(true)
 		assert.NoError(t, tableData.MergeColumnsFromDestination(backfilledCol))
 		for _, inMemoryCol := range tableData.inMemoryColumns.GetColumns() {
-			if inMemoryCol.RawName() == backfilledCol.RawName() {
-				assert.True(t, inMemoryCol.Backfilled(), inMemoryCol.RawName())
+			if inMemoryCol.Name() == backfilledCol.Name() {
+				assert.True(t, inMemoryCol.Backfilled(), inMemoryCol.Name())
 			} else {
-				assert.False(t, inMemoryCol.Backfilled(), inMemoryCol.RawName())
+				assert.False(t, inMemoryCol.Backfilled(), inMemoryCol.Name())
 			}
 		}
 
diff --git a/lib/optimization/table_data.go b/lib/optimization/table_data.go
index 42338aa5f..09017077e 100644
--- a/lib/optimization/table_data.go
+++ b/lib/optimization/table_data.go
@@ -257,9 +257,9 @@ func (t *TableData) MergeColumnsFromDestination(destCols ...columns.Column) erro
 		var foundColumn columns.Column
 		var found bool
 		for _, destCol := range destCols {
-			if destCol.RawName() == strings.ToLower(inMemoryCol.RawName()) {
+			if destCol.Name() == strings.ToLower(inMemoryCol.Name()) {
 				if destCol.KindDetails.Kind == typing.Invalid.Kind {
-					return fmt.Errorf("column %q is invalid", destCol.RawName())
+					return fmt.Errorf("column %q is invalid", destCol.Name())
 				}
 
 				foundColumn = destCol
diff --git a/lib/optimization/table_data_test.go b/lib/optimization/table_data_test.go
index 74cea7f52..5a4e8f449 100644
--- a/lib/optimization/table_data_test.go
+++ b/lib/optimization/table_data_test.go
@@ -145,7 +145,7 @@ func TestTableData_UpdateInMemoryColumns(t *testing.T) {
 	assert.True(t, isOk)
 
 	extCol.KindDetails.ExtendedTimeDetails.Format = time.RFC3339Nano
-	tableData.inMemoryColumns.UpdateColumn(columns.NewColumn(extCol.RawName(), extCol.KindDetails))
+	tableData.inMemoryColumns.UpdateColumn(columns.NewColumn(extCol.Name(), extCol.KindDetails))
 
 	for name, colKindDetails := range map[string]typing.KindDetails{
 		"foo":                  typing.String,
diff --git a/lib/parquetutil/generate_schema.go b/lib/parquetutil/generate_schema.go
index dfbbf1490..8f02657d2 100644
--- a/lib/parquetutil/generate_schema.go
+++ b/lib/parquetutil/generate_schema.go
@@ -12,7 +12,7 @@ func GenerateJSONSchema(columns []columns.Column) (string, error) {
 	var fields []typing.Field
 	for _, column := range columns {
 		// We don't need to escape the column name here.
-		field, err := column.KindDetails.ParquetAnnotation(column.RawName())
+		field, err := column.KindDetails.ParquetAnnotation(column.Name())
 		if err != nil {
 			return "", err
 		}
diff --git a/lib/typing/columns/columns.go b/lib/typing/columns/columns.go
index 23eab0705..c7c71053a 100644
--- a/lib/typing/columns/columns.go
+++ b/lib/typing/columns/columns.go
@@ -79,7 +79,7 @@ func (c *Column) ShouldBackfill() bool {
 	return c.defaultValue != nil && !c.backfilled
 }
 
-func (c *Column) RawName() string {
+func (c *Column) Name() string {
 	return c.name
 }
 
@@ -187,7 +187,7 @@ func (c *Columns) GetColumnsToUpdate() []string {
 			continue
 		}
 
-		cols = append(cols, col.RawName())
+		cols = append(cols, col.Name())
 	}
 
 	return cols
@@ -240,7 +240,7 @@ func (c *Columns) UpdateQuery(dialect sql.Dialect, skipDeleteCol bool) string {
 		}
 
 		// skipDeleteCol is useful because we don't want to copy the deleted column over to the source table if we're doing a hard row delete.
-		if skipDeleteCol && column.RawName() == constants.DeleteColumnMarker {
+		if skipDeleteCol && column.Name() == constants.DeleteColumnMarker {
 			continue
 		}
 
diff --git a/lib/typing/columns/columns_test.go b/lib/typing/columns/columns_test.go
index 4610977c5..525ee12e3 100644
--- a/lib/typing/columns/columns_test.go
+++ b/lib/typing/columns/columns_test.go
@@ -168,7 +168,7 @@ func TestColumn_Name(t *testing.T) {
 			name: testCase.colName,
 		}
 
-		assert.Equal(t, testCase.expectedName, col.RawName(), testCase.colName)
+		assert.Equal(t, testCase.expectedName, col.Name(), testCase.colName)
 
 		assert.Equal(t, testCase.expectedNameEsc, col.EscapedName(sql.SnowflakeDialect{}), testCase.colName)
 		assert.Equal(t, testCase.expectedNameEscBq, col.EscapedName(sql.BigQueryDialect{}), testCase.colName)
diff --git a/lib/typing/columns/diff.go b/lib/typing/columns/diff.go
index ac22c27d5..f717bca1f 100644
--- a/lib/typing/columns/diff.go
+++ b/lib/typing/columns/diff.go
@@ -40,7 +40,7 @@ func Diff(columnsInSource *Columns, columnsInDestination *Columns, softDelete bo
 	targ := CloneColumns(columnsInDestination)
 	var colsToDelete []Column
 	for _, col := range src.GetColumns() {
-		_, isOk := targ.GetColumn(col.RawName())
+		_, isOk := targ.GetColumn(col.Name())
 		if isOk {
 			colsToDelete = append(colsToDelete, col)
 
@@ -49,13 +49,13 @@ func Diff(columnsInSource *Columns, columnsInDestination *Columns, softDelete bo
 
 	// We cannot delete inside a for-loop that is iterating over src.GetColumns() because we are messing up the array order.
 	for _, colToDelete := range colsToDelete {
-		src.DeleteColumn(colToDelete.RawName())
-		targ.DeleteColumn(colToDelete.RawName())
+		src.DeleteColumn(colToDelete.Name())
+		targ.DeleteColumn(colToDelete.Name())
 	}
 
 	var targetColumnsMissing Columns
 	for _, col := range src.GetColumns() {
-		if shouldSkipColumn(col.RawName(), softDelete, includeArtieUpdatedAt, includeDatabaseUpdatedAt, mode) {
+		if shouldSkipColumn(col.Name(), softDelete, includeArtieUpdatedAt, includeDatabaseUpdatedAt, mode) {
 			continue
 		}
 
@@ -64,7 +64,7 @@ func Diff(columnsInSource *Columns, columnsInDestination *Columns, softDelete bo
 
 	var sourceColumnsMissing Columns
 	for _, col := range targ.GetColumns() {
-		if shouldSkipColumn(col.RawName(), softDelete, includeArtieUpdatedAt, includeDatabaseUpdatedAt, mode) {
+		if shouldSkipColumn(col.Name(), softDelete, includeArtieUpdatedAt, includeDatabaseUpdatedAt, mode) {
 			continue
 		}
 
diff --git a/lib/typing/columns/diff_test.go b/lib/typing/columns/diff_test.go
index 93a24e90e..8609cedeb 100644
--- a/lib/typing/columns/diff_test.go
+++ b/lib/typing/columns/diff_test.go
@@ -226,7 +226,7 @@ func TestDiffDeterministic(t *testing.T) {
 
 		var key string
 		for _, targetKeyMissing := range targetKeysMissing {
-			key += targetKeyMissing.RawName()
+			key += targetKeyMissing.Name()
 		}
 
 		retMap[key] = false
diff --git a/models/event/event_save_test.go b/models/event/event_save_test.go
index d37b7f2d0..6c7c98f01 100644
--- a/models/event/event_save_test.go
+++ b/models/event/event_save_test.go
@@ -48,7 +48,7 @@ func (e *EventsTestSuite) TestSaveEvent() {
 	// Check the in-memory DB columns.
 	var found int
 	for _, col := range optimization.ReadOnlyInMemoryCols().GetColumns() {
-		if col.RawName() == expectedLowerCol || col.RawName() == anotherLowerCol {
+		if col.Name() == expectedLowerCol || col.Name() == anotherLowerCol {
 			found += 1
 		}
 
@@ -183,16 +183,16 @@ func (e *EventsTestSuite) TestEvent_SaveColumnsNoData() {
 	td := e.db.GetOrCreateTableData("non_existent")
 	var prevKey string
 	for _, col := range td.ReadOnlyInMemoryCols().GetColumns() {
-		if col.RawName() == constants.DeleteColumnMarker {
+		if col.Name() == constants.DeleteColumnMarker {
 			continue
 		}
 
 		if prevKey == "" {
-			prevKey = col.RawName()
+			prevKey = col.Name()
 			continue
 		}
 
-		currentKeyParsed, err := strconv.Atoi(col.RawName())
+		currentKeyParsed, err := strconv.Atoi(col.Name())
 		assert.NoError(e.T(), err)
 
 		prevKeyParsed, err := strconv.Atoi(prevKey)
@@ -206,7 +206,7 @@ func (e *EventsTestSuite) TestEvent_SaveColumnsNoData() {
 	evt.Columns.AddColumn(columns.NewColumn("foo", typing.Invalid))
 	var index int
 	for idx, col := range evt.Columns.GetColumns() {
-		if col.RawName() == "foo" {
+		if col.Name() == "foo" {
 			index = idx
 		}
 	}